Class: Dtn::Streaming::Client

Inherits:
Object
  • Object
show all
Includes:
Status
Defined in:
lib/dtn/streaming/client.rb

Overview

Top level client abstraction. Different streams are available on different ports, so we can use it and follow the same pattern

Defined Under Namespace

Modules: Status

Constant Summary collapse

PROTOCOL_VERSION = 
"6.1"
CLIENT_TERMINATION_SIGNALS = 
%w[TERM INT].freeze
COMMON_SUPPORTED_MESSAGES = 
{
  "S" => Messages::System::Generic,
  "T" => Messages::System::Timestamp,
  "n" => Messages::System::SymbolNotFound,
  "E" => Messages::System::Error
}.freeze

Constants included from Status

Status::STATUSES

Instance Attribute Summary collapse

Attributes included from Status

#status

Instance Method Summary collapse

Constructor Details

#initialize(name: nil, start_engine: true) ⇒ Client

Returns a new instance of Client.



39
40
41
42
43
44
45
46
47
 
# File 'lib/dtn/streaming/client.rb', line 39

def initialize(name: nil, start_engine: true)
  @name = name || SecureRandom.alphanumeric(10)

  initializing

  init_connection
  setup_signals
  engine if start_engine
end

Instance Attribute Details

#name ⇒ Object (readonly)

Returns the value of attribute name.



49
50
51
 
# File 'lib/dtn/streaming/client.rb', line 49

def name
  @name
end

#quote_update_fields ⇒ Object

We are able to filer the incoming data with custom fields using



53
54
55
 
# File 'lib/dtn/streaming/client.rb', line 53

def quote_update_fields
  @quote_update_fields
end

Instance Method Details

#engine ⇒ Object 



71
72
73
74
75
76
77
78
 
# File 'lib/dtn/streaming/client.rb', line 71

def engine
  @engine ||= Thread.new do
    run
    while running? && (line = socket.gets)
      process_line(line: line)
    end
  end
end

#observers ⇒ Object 



59
60
61
 
# File 'lib/dtn/streaming/client.rb', line 59

def observers
  @observers ||= Set.new
end

#request ⇒ Object 



55
56
57
 
# File 'lib/dtn/streaming/client.rb', line 55

def request
  RequestBuilder.new(client: self)
end

#socket ⇒ Object 



67
68
69
 
# File 'lib/dtn/streaming/client.rb', line 67

def socket
  @socket ||= TCPSocket.open(Dtn.host, self.class::PORT)
end

#to_s ⇒ Object 



63
64
65
 
# File 'lib/dtn/streaming/client.rb', line 63

def to_s
  "Client name: #{name}, status: #{status}"
end