Module: OMQ::Transport::Lz4Tcp

Defined in:
lib/omq/transport/lz4_tcp/transport.rb,
lib/omq/transport/lz4_tcp/connection.rb

Defined Under Namespace

Classes: Dialer, Listener, Lz4Connection

Constant Summary collapse

SCHEME = 
"lz4+tcp"

Class Method Summary collapse

Class Method Details

.connect_timeout(options) ⇒ Object 



104
105
106
107
108
 
# File 'lib/omq/transport/lz4_tcp/transport.rb', line 104

def connect_timeout(options)
  ri = options.reconnect_interval
  ri = ri.end if ri.is_a?(Range)
  [ri, 0.5].max
end

.connection_class ⇒ Object

Called by OMQ::Engine::ConnectionLifecycle after the ZMTP handshake completes; we return the default connection class so that handshake itself runs uncompressed over raw TCP.



19
20
21
 
# File 'lib/omq/transport/lz4_tcp/transport.rb', line 19

def connection_class
  Protocol::ZMTP::Connection
end

.dialer(endpoint, engine, dict: nil, auto_dict: nil) ⇒ Dialer

Creates an lz4+tcp dialer for an endpoint.

Parameters:

  • endpoint (String)
  • engine (OMQ::Engine)
  • dict (String, nil) (defaults to: nil) —

    user-supplied dictionary bytes.

  • auto_dict (true, Hash, nil) (defaults to: nil) —

    enable automatic dictionary training. See listener for details.

Returns:



67
68
69
70
71
 
# File 'lib/omq/transport/lz4_tcp/transport.rb', line 67

def dialer(endpoint, engine, dict: nil, auto_dict: nil, **)
  validate_dict!(dict)
  validate_auto_dict!(auto_dict, dict)
  Dialer.new(endpoint, engine, dict&.b, normalize_auto_dict(auto_dict))
end

.listener(endpoint, engine, dict: nil, auto_dict: nil) ⇒ Listener

Creates a bound lz4+tcp listener.

Parameters:

  • endpoint (String) —

    e.g. "lz4+tcp://127.0.0.1:0"

  • engine (OMQ::Engine)
  • dict (String, nil) (defaults to: nil) —

    user-supplied dictionary bytes to ship on the first outgoing message. If nil, no dict is shipped (payloads compress without dict or go plaintext below the min-size threshold).

  • auto_dict (true, Hash, nil) (defaults to: nil) —

    enable automatic dictionary training from early traffic (RFC ยง7.6). Pass true for defaults (capacity=2048, trigger=100) or a Hash with :capacity and/or :trigger keys. Mutually exclusive with dict:.

Returns:



38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
 
# File 'lib/omq/transport/lz4_tcp/transport.rb', line 38

def listener(endpoint, engine, dict: nil, auto_dict: nil, **)
  validate_dict!(dict)
  validate_auto_dict!(auto_dict, dict)

  host, port  = parse_endpoint(endpoint)
  lookup_host = normalize_bind_host(host)
  servers     = ::Socket.tcp_server_sockets(lookup_host, port)

  if servers.empty?
    raise ::Socket::ResolutionError, "no addresses for #{host.inspect}"
  end

  actual_port  = servers.first.local_address.ip_port
  display_host = host == "*" ? "*" : (lookup_host || "*")
  host_part    = display_host.include?(":") ? "[#{display_host}]" : display_host
  resolved     = "#{SCHEME}://#{host_part}:#{actual_port}"

  Listener.new(resolved, servers, actual_port, engine, dict&.b, normalize_auto_dict(auto_dict))
end

.normalize_bind_host(host) ⇒ Object 



87
88
89
90
91
92
93
 
# File 'lib/omq/transport/lz4_tcp/transport.rb', line 87

def normalize_bind_host(host)
  case host
  when "*" then nil
  when nil, "", "localhost" then TCP.loopback_host
  else host
  end
end

.normalize_connect_host(host) ⇒ Object 



96
97
98
99
100
101
 
# File 'lib/omq/transport/lz4_tcp/transport.rb', line 96

def normalize_connect_host(host)
  case host
  when nil, "", "*", "localhost" then TCP.loopback_host
  else host
  end
end

.parse_endpoint(endpoint) ⇒ Object 



81
82
83
84
 
# File 'lib/omq/transport/lz4_tcp/transport.rb', line 81

def parse_endpoint(endpoint)
  uri = URI.parse(endpoint)
  [uri.hostname, uri.port]
end

.validate_endpoint!(endpoint) ⇒ Object 



74
75
76
77
78
 
# File 'lib/omq/transport/lz4_tcp/transport.rb', line 74

def validate_endpoint!(endpoint)
  host, _port = parse_endpoint(endpoint)
  lookup_host = normalize_connect_host(host)
  Addrinfo.getaddrinfo(lookup_host, nil, nil, :STREAM) if lookup_host
end