Class: OMQ::Transport::Lz4Tcp::Lz4Connection

Inherits:
SimpleDelegator
  • Object
show all
Defined in:
lib/omq/transport/lz4_tcp/connection.rb

Overview

Per-connection state + encode/decode hooks. A SimpleDelegator over the ZMTP connection so send_message / write_message / receive_message route through compression, but everything else (write_command, close, etc.) passes through untouched.

Constant Summary collapse

DEFAULT_DICT_CAPACITY = 
2048
DEFAULT_TRAIN_TRIGGER = 
100

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(conn, send_dict_bytes:, max_message_size:, auto_dict: nil) ⇒ Lz4Connection

Returns a new instance of Lz4Connection.



22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
 
# File 'lib/omq/transport/lz4_tcp/connection.rb', line 22

def initialize(conn, send_dict_bytes:, max_message_size:, auto_dict: nil)
  super(conn)
  @max_message_size = max_message_size

  @recv_codec       = build_block_codec(nil)
  @recv_dict_bytes  = nil
  @last_wire_size_in = nil

  if send_dict_bytes
    @send_dict_bytes   = send_dict_bytes.b
    @send_codec        = build_block_codec(@send_dict_bytes)
    @send_dict_shipped = false
    @trainer           = nil
  elsif auto_dict
    @send_dict_bytes   = nil
    @send_codec        = build_block_codec(nil)
    @send_dict_shipped = true
    @trainer           = Lz4rip::DictTrainer.new(auto_dict[:capacity])
    @train_trigger     = auto_dict[:trigger]
    @train_msg_count   = 0
  else
    @send_dict_bytes   = nil
    @send_codec        = build_block_codec(nil)
    @send_dict_shipped = true
    @trainer           = nil
  end
end

Instance Attribute Details

#last_wire_size_in ⇒ Integer? (readonly)

Returns wire bytesize of the last received message (sum across parts, compressed sentinel included).

Returns:

  • (Integer, nil) —

    wire bytesize of the last received message (sum across parts, compressed sentinel included).



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

def last_wire_size_in
  @last_wire_size_in
end

Instance Method Details

#receive_message ⇒ Object 



75
76
77
78
79
80
81
82
83
84
85
86
87
 
# File 'lib/omq/transport/lz4_tcp/connection.rb', line 75

def receive_message
  # Loop: a dict shipment is consumed silently and we read the
  # next ZMTP message. Only data messages are returned to the
  # caller. Budget tracking happens inside decode_wire_parts.
  loop do
    parts   = __getobj__.receive_message
    decoded = decode_wire_parts(parts)
    if decoded
      @last_wire_size_in = parts.sum(&:bytesize)
      return decoded
    end
  end
end

#send_message(parts) ⇒ Object 



51
52
53
54
55
56
 
# File 'lib/omq/transport/lz4_tcp/connection.rb', line 51

def send_message(parts)
  maybe_train!(parts)
  wire = encode_parts(parts)
  ship_send_dict!
  __getobj__.send_message(wire)
end

#write_message(parts) ⇒ Object 



59
60
61
62
63
64
 
# File 'lib/omq/transport/lz4_tcp/connection.rb', line 59

def write_message(parts)
  maybe_train!(parts)
  wire = encode_parts(parts)
  ship_send_dict!
  __getobj__.write_message(wire)
end

#write_messages(messages) ⇒ Object 



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

def write_messages(messages)
  messages.each { |parts| maybe_train!(parts) }
  wires = messages.map { |parts| encode_parts(parts) }
  ship_send_dict!
  __getobj__.write_messages(wires)
end