Class: Protocol::ZMTP::Codec::Command

Inherits:
Object
  • Object
show all
Defined in:
lib/protocol/zmtp/codec/command.rb

Overview

ZMTP command encode/decode.

Command frame body format:

1 byte:    command name length
N bytes:   command name
remaining: command data

READY command data = property list:

1 byte:  property name length
N bytes: property name
4 bytes: property value length (big-endian)
N bytes: property value

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(name, data = EMPTY_BINARY) ⇒ Command

Returns a new instance of Command.

Parameters:

  • name (String) —

    command name

  • data (String) (defaults to: EMPTY_BINARY) —

    command data



30
31
32
33
 
# File 'lib/protocol/zmtp/codec/command.rb', line 30

def initialize(name, data = EMPTY_BINARY)
  @name = name
  @data = data.encoding == Encoding::BINARY ? data : data.b
end

Instance Attribute Details

#data ⇒ String (readonly)

Returns command data (binary).

Returns:

  • (String) —

    command data (binary)



25
26
27
 
# File 'lib/protocol/zmtp/codec/command.rb', line 25

def data
  @data
end

#name ⇒ String (readonly)

Returns command name (e.g. “READY”, “SUBSCRIBE”).

Returns:

  • (String) —

    command name (e.g. “READY”, “SUBSCRIBE”)



21
22
23
 
# File 'lib/protocol/zmtp/codec/command.rb', line 21

def name
  @name
end

Class Method Details

.cancel(prefix) ⇒ Command

Builds a CANCEL command (unsubscribe).

Parameters:

  • prefix (String) —

    subscription prefix to cancel

Returns:



105
106
107
 
# File 'lib/protocol/zmtp/codec/command.rb', line 105

def self.cancel(prefix)
  new("CANCEL", prefix.b)
end

.decode_properties(data) ⇒ Hash{String => String}

Decodes a ZMTP property list from binary data.

Parameters:

  • data (String) —

    binary-encoded property list

Returns:

  • (Hash{String => String}) —

    property name-value pairs

Raises:

  • (Error) —

    on malformed property data



185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
 
# File 'lib/protocol/zmtp/codec/command.rb', line 185

def self.decode_properties(data)
  result = {}
  offset = 0

  while offset < data.bytesize
    raise Error, "property name truncated" if offset + 1 > data.bytesize
    name_len = data.getbyte(offset)
    offset += 1

    raise Error, "property name truncated" if offset + name_len > data.bytesize
    name = data.byteslice(offset, name_len)
    offset += name_len

    raise Error, "property value length truncated" if offset + 4 > data.bytesize
    value_len = data.byteslice(offset, 4).unpack1("N")
    offset += 4

    raise Error, "property value truncated" if offset + value_len > data.bytesize
    value = data.byteslice(offset, value_len)
    offset += value_len

    result[name] = value
  end

  result
end

.encode_properties(props) ⇒ String

Encodes a hash of properties into ZMTP property list format.

Parameters:

  • props (Hash{String => String}) —

    property name-value pairs

Returns:

  • (String) —

    binary-encoded property list



170
171
172
173
174
175
176
177
 
# File 'lib/protocol/zmtp/codec/command.rb', line 170

def self.encode_properties(props)
  parts = props.map do |name, value|
    name_bytes  = name.b
    value_bytes = value.b
    name_bytes.bytesize.chr.b + name_bytes + [value_bytes.bytesize].pack("N") + value_bytes
  end
  parts.join
end

.from_body(body) ⇒ Command

Decodes a command from a frame body.

Parameters:

  • body (String) —

    binary frame body

Returns:

Raises:

  • (Error) —

    on malformed command



59
60
61
62
63
64
65
66
67
68
69
70
 
# File 'lib/protocol/zmtp/codec/command.rb', line 59

def self.from_body(body)
  body = body.b
  raise Error, "command body too short" if body.bytesize < 1

  name_len = body.getbyte(0)

  raise Error, "command name truncated" if body.bytesize < 1 + name_len

  name = body.byteslice(1, name_len)
  data = body.byteslice(1 + name_len..)
  new(name, data)
end

.join(group) ⇒ Command

Builds a JOIN command (RADIO/DISH group subscription).

Parameters:

  • group (String) —

    group name to join

Returns:



114
115
116
 
# File 'lib/protocol/zmtp/codec/command.rb', line 114

def self.join(group)
  new("JOIN", group.b)
end

.leave(group) ⇒ Command

Builds a LEAVE command (RADIO/DISH group unsubscription).

Parameters:

  • group (String) —

    group name to leave

Returns:



123
124
125
 
# File 'lib/protocol/zmtp/codec/command.rb', line 123

def self.leave(group)
  new("LEAVE", group.b)
end

.ping(ttl: 0, context: "".b) ⇒ Command

Builds a PING command.

Parameters:

  • ttl (Numeric) (defaults to: 0) —

    time-to-live in seconds (sent as deciseconds)

  • context (String) (defaults to: "".b) —

    optional context bytes (up to 16 bytes)

Returns:



133
134
135
136
 
# File 'lib/protocol/zmtp/codec/command.rb', line 133

def self.ping(ttl: 0, context: "".b)
  ttl_ds = (ttl * 10).to_i
  new("PING", [ttl_ds].pack("n") + context.b)
end

.pong(context: "".b) ⇒ Command

Builds a PONG command.

Parameters:

  • context (String) (defaults to: "".b) —

    context bytes echoed from the PING

Returns:



143
144
145
 
# File 'lib/protocol/zmtp/codec/command.rb', line 143

def self.pong(context: "".b)
  new("PONG", context.b)
end

.ready(socket_type:, identity: "", qos: 0, qos_hash: "", metadata: nil) ⇒ Command

Builds a READY command with Socket-Type, Identity, optional X-QoS, and any extra properties supplied by upper layers (e.g. an extension that injects an ‘X-Compression` property).

Parameters:

  • qos (Integer) (defaults to: 0) —

    QoS level (0 = omitted)

  • qos_hash (String) (defaults to: "") —

    supported hash algorithms in preference order (e.g. “xXsS”)

  • metadata (Hash{String => String}) (defaults to: nil) —

    additional READY properties

Returns:



81
82
83
84
85
86
87
88
89
 
# File 'lib/protocol/zmtp/codec/command.rb', line 81

def self.ready(socket_type:, identity: "", qos: 0, qos_hash: "", metadata: nil)
  props = { "Socket-Type" => socket_type, "Identity" => identity }
  if qos > 0
    props["X-QoS"]      = qos.to_s
    props["X-QoS-Hash"] = qos_hash unless qos_hash.empty?
  end
  props.merge!(metadata) if metadata && !metadata.empty?
  new("READY", encode_properties(props))
end

.subscribe(prefix) ⇒ Command

Builds a SUBSCRIBE command.

Parameters:

  • prefix (String) —

    subscription prefix to match

Returns:



96
97
98
 
# File 'lib/protocol/zmtp/codec/command.rb', line 96

def self.subscribe(prefix)
  new("SUBSCRIBE", prefix.b)
end

Instance Method Details

#ping_ttl_and_context ⇒ Array(Numeric, String)

Extracts TTL (in seconds) and context from a PING command’s data.

Returns:

  • (Array(Numeric, String)) —
    ttl_seconds, context_bytes


151
152
153
154
155
 
# File 'lib/protocol/zmtp/codec/command.rb', line 151

def ping_ttl_and_context
  ttl_ds  = @data.unpack1("n")
  context = @data.bytesize > 2 ? @data.byteslice(2..) : "".b
  [ttl_ds / 10.0, context]
end

#properties ⇒ Hash{String => String}

Parses READY command data as a property list.

Returns:

  • (Hash{String => String}) —

    property name-value pairs



161
162
163
 
# File 'lib/protocol/zmtp/codec/command.rb', line 161

def properties
  self.class.decode_properties(@data)
end

#to_body ⇒ String

Encodes as a command frame body.

Returns:

  • (String) —

    binary body (name-length + name + data)



39
40
41
42
43
 
# File 'lib/protocol/zmtp/codec/command.rb', line 39

def to_body
  name_bytes = @name.encoding == Encoding::BINARY ? @name : @name.b
  buf = String.new(capacity: 1 + name_bytes.bytesize + @data.bytesize, encoding: Encoding::BINARY)
  buf << Frame::FLAG_BYTES[name_bytes.bytesize] << name_bytes << @data
end

#to_frame ⇒ Frame

Encodes as a complete command Frame.

Returns:



49
50
51
 
# File 'lib/protocol/zmtp/codec/command.rb', line 49

def to_frame
  Frame.new(to_body, command: true)
end