Class: Mfp::Client

Inherits:
Object
  • Object
show all
Defined in:
lib/mfp/client.rb

Defined Under Namespace

Classes: Settings

Constant Summary collapse

DEFAULT_OPTS = 
{
  ca_certs: [],
  key: nil,
  cert: nil,
  insecure: false,
  application: nil,
  hostname: Socket.gethostname,
  disable_compression: false,
  idle_timeout_secs: 30,
  ping_timeout_secs: 10,
  logger: Logrb.noop,
}.freeze

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(host, port, call_sign) ⇒ Client

Returns a new instance of Client.

Raises:



33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
 
# File 'lib/mfp/client.rb', line 33

def initialize(host, port, call_sign, **)
  raise InvalidCallSignLengthError, "call_sign must have 4 bytes" if call_sign.bytesize != 4

  # State

  @running = true
  @to_write = Async::Queue.new
  @to_read = Async::Queue.new
  @call_sign = call_sign
  @max_streams = nil
  # this is set initially to zero so we can transmit the HELLO payload.
  # Once the handshake is completed, this value will be replaced with
  # whatever the server returns to us.
  @max_payload = 0
  @inner_tasks = []

  # Networking

  @host = host
  @port = port
  @settings = Settings.new(**DEFAULT_OPTS.dup, **)

  @logger = @settings.logger

  @endpoint = if (@settings.cert && @settings.key) || !@settings.ca_certs.empty? || @settings.insecure
    @logger.debug("Starting TLS client", hostname: host, port:)
    ctx = Mfp.default_tls_context
    ctx.cert = OpenSSL::X509::Certificate.new(@settings.cert) if @settings.cert
    ctx.key = OpenSSL::PKey.read(@settings.key) if @settings.key
    ctx.verify_mode = @settings.insecure ? OpenSSL::SSL::VERIFY_NONE : OpenSSL::SSL::VERIFY_PEER
    store = OpenSSL::X509::Store.new
    store.set_default_paths
    @settings.ca_certs.each do |raw|
      store.add_cert(OpenSSL::X509::Certificate.new(raw))
    end
    ctx.cert_store = store
    IO::Endpoint.ssl(host, port, ssl_context: ctx)
  else
    @logger.debug("Starting plain client", hostname: host, port:)
    IO::Endpoint.tcp(host, port)
  end

  @compressor = Compressor.new(:identity)
  @ping_handler = PingHandler.new(
    idle_timeout: @settings.idle_timeout_secs,
    ping_timeout: @settings.ping_timeout_secs,
  )
end

Instance Attribute Details

#max_payload ⇒ Object (readonly)

Returns the value of attribute max_payload.



31
32
33
 
# File 'lib/mfp/client.rb', line 31

def max_payload
  @max_payload
end

#max_streams ⇒ Object (readonly)

Returns the value of attribute max_streams.



31
32
33
 
# File 'lib/mfp/client.rb', line 31

def max_streams
  @max_streams
end

Instance Method Details

#close ⇒ Object 



101
102
103
104
105
106
 
# File 'lib/mfp/client.rb', line 101

def close
  return unless @running

  abort!(reason: CONNECTION_EVENT_CODE_DISCONNECTED)
  nil
end

#recv ⇒ Object 



108
 
# File 'lib/mfp/client.rb', line 108

def recv = @to_read.dequeue

#send_control(id, msg, flags: 0) ⇒ Object 



120
121
122
123
124
125
126
127
128
 
# File 'lib/mfp/client.rb', line 120

def send_control(id, msg, flags: 0)
  send(Proto::Frame.new(
    stream_id: id,
    kind: Proto::MessageKind::CONTROL,
    flags:,
    payload: msg,
    length: msg&.bytesize || 0,
  ))
end

#send_stream(id, msg, flags: 0) ⇒ Object 



110
111
112
113
114
115
116
117
118
 
# File 'lib/mfp/client.rb', line 110

def send_stream(id, msg, flags: 0)
  send(Proto::Frame.new(
    stream_id: id,
    kind: Proto::MessageKind::STREAM,
    flags:,
    payload: msg,
    length: msg&.bytesize || 0,
  ))
end

#start(task) ⇒ Object 



82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
 
# File 'lib/mfp/client.rb', line 82

def start(task)
  @task = task.async do |t|
    @socket = @endpoint.connect
    @inner_tasks << t.async { handle_writes }
    setup
  end

  begin
    @task.wait
  rescue StandardError => e
    stop_inner_tasks!
    raise Mfp::Error, "handshake failed: #{e.message}"
  end

  @inner_tasks << task.async { handle_reads }
  @inner_tasks << task.async { @ping_handler.start(it) }
  @inner_tasks << task.async { handle_pings }
end