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
 
# File 'lib/omq/transport/zstd_tcp/connection.rb', line 14

def initialize(conn, codec)
  super(conn)
  @codec              = codec
  @dict_shipped       = false
  @recv_dict          = 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 



44
45
46
47
48
49
50
51
52
53
 
# File 'lib/omq/transport/zstd_tcp/connection.rb', line 44

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) 


56
57
58
59
 
# File 'lib/omq/transport/zstd_tcp/connection.rb', line 56

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

#send_message(parts) ⇒ Object 



23
24
25
26
27
 
# File 'lib/omq/transport/zstd_tcp/connection.rb', line 23

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

#write_message(parts) ⇒ Object 



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

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

#write_messages(messages) ⇒ Object 



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

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