Class: OMQ::Transport::ZstdTcp::ZstdConnection

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

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(conn, codec) ⇒ ZstdConnection

Returns a new instance of ZstdConnection.



14
15
16
17
18
19
20
21
22
23
24
25
 
# File 'lib/omq/transport/zstd_tcp/connection.rb', line 14

def initialize(conn, codec)
  super(conn)
  @codec              = codec
  @dict_shipped       = false
  # FrameCodec is the decoder. Starts no-dict; when a dict
  # shipment arrives on this direction, we build a fresh
  # dict-bound FrameCodec and replace this one.
  @recv_codec         = Zrip::FrameCodec.new
  @recv_no_dict_codec = @recv_codec
  @recv_dict_bytes    = nil
  @last_wire_size_in  = nil
end

Instance Attribute Details

#last_wire_size_in ⇒ Integer? (readonly)

Returns wire bytesize of the last received message.

Returns:

  • (Integer, nil) —

    wire bytesize of the last received message



11
12
13
 
# File 'lib/omq/transport/zstd_tcp/connection.rb', line 11

def last_wire_size_in
  @last_wire_size_in
end

Instance Method Details

#receive_message ⇒ Object 



49
50
51
52
53
54
55
56
57
58
 
# File 'lib/omq/transport/zstd_tcp/connection.rb', line 49

def receive_message
  loop do
    parts   = __getobj__.receive_message
    decoded = decode_parts(parts)
    if decoded
      @last_wire_size_in = parts.sum { |p| p.bytesize }
      return decoded
    end
  end
end

#respond_to?(name, include_private = false) ⇒ Boolean

Returns:

  • (Boolean) 


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

def respond_to?(name, include_private = false)
  return false if name == :write_wire
  super
end

#send_message(parts) ⇒ Object 



28
29
30
31
32
 
# File 'lib/omq/transport/zstd_tcp/connection.rb', line 28

def send_message(parts)
  compressed = @codec.compress_parts(parts)
  ship_dict!
  __getobj__.send_message(compressed)
end

#write_message(parts) ⇒ Object 



35
36
37
38
39
 
# File 'lib/omq/transport/zstd_tcp/connection.rb', line 35

def write_message(parts)
  compressed = @codec.compress_parts(parts)
  ship_dict!
  __getobj__.write_message(compressed)
end

#write_messages(messages) ⇒ Object 



42
43
44
45
46
 
# File 'lib/omq/transport/zstd_tcp/connection.rb', line 42

def write_messages(messages)
  compressed = messages.map { |parts| @codec.compress_parts(parts) }
  ship_dict!
  __getobj__.write_messages(compressed)
end