Class: OMQ::Transport::ZstdTcp::Codec

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

Constant Summary collapse

MAX_DICT_SIZE = 
64 * 1024
DICT_CAPACITY = 
8 * 1024
TRAIN_MAX_SAMPLES = 
1000
TRAIN_MAX_BYTES = 
100 * 1024
TRAIN_MAX_SAMPLE_LEN = 
1024
MIN_COMPRESS_NO_DICT = 
512
MIN_COMPRESS_WITH_DICT = 
64
NUL_PREAMBLE = 
("\x00" * 4).b.freeze
ZSTD_MAGIC = 
"\x28\xB5\x2F\xFD".b.freeze
ZDICT_MAGIC = 
"\x37\xA4\x30\xEC".b.freeze
USER_DICT_ID_RANGE = 
(32_768..(2**31 - 1)).freeze

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(level:, dict: nil, max_message_size: nil) ⇒ Codec

Returns a new instance of Codec.



25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
 
# File 'lib/omq/transport/zstd_tcp/codec.rb', line 25

def initialize(level:, dict: nil, max_message_size: nil)
  @level            = level
  @max_message_size = max_message_size

  @send_dict       = nil
  @send_dict_bytes = nil

  @training      = dict.nil?
  @train_samples = []
  @train_bytes   = 0

  @cached_parts      = nil
  @cached_compressed = nil

  install_send_dict(dict.b) if dict
end

Instance Attribute Details

#max_message_size ⇒ Object (readonly)

Returns the value of attribute max_message_size.



22
23
24
 
# File 'lib/omq/transport/zstd_tcp/codec.rb', line 22

def max_message_size
  @max_message_size
end

#send_dict_bytes ⇒ Object (readonly)

Returns the value of attribute send_dict_bytes.



22
23
24
 
# File 'lib/omq/transport/zstd_tcp/codec.rb', line 22

def send_dict_bytes
  @send_dict_bytes
end

Instance Method Details

#compress_parts(parts) ⇒ Object 



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

def compress_parts(parts)
  return @cached_compressed if parts.equal?(@cached_parts)

  parts.each { |p| maybe_train!(p) }

  compressed = parts.map { |p| compress_or_plain(p) }
  @cached_parts      = parts
  @cached_compressed = compressed.freeze
  compressed
end

#parse_frame_content_size(wire) ⇒ Object 



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
81
82
 
# File 'lib/omq/transport/zstd_tcp/codec.rb', line 55

def parse_frame_content_size(wire)
  return nil if wire.bytesize < 5

  fhd        = wire.getbyte(4)
  did_flag   = fhd & 0x03
  single_seg = (fhd >> 5) & 0x01
  fcs_flag   = (fhd >> 6) & 0x03

  return nil if fcs_flag == 0 && single_seg == 0

  off = 5 + (single_seg == 0 ? 1 : 0) + [0, 1, 2, 4][did_flag]

  case fcs_flag
  when 0
    return nil if wire.bytesize < off + 1
    wire.getbyte(off)
  when 1
    return nil if wire.bytesize < off + 2
    wire.byteslice(off, 2).unpack1("v") + 256
  when 2
    return nil if wire.bytesize < off + 4
    wire.byteslice(off, 4).unpack1("V")
  when 3
    return nil if wire.bytesize < off + 8
    lo, hi = wire.byteslice(off, 8).unpack("VV")
    (hi << 32) | lo
  end
end