Class: Krakow::ConnectionFeatures::SnappyFrames::Io

Inherits:
Object
  • Object
show all
Defined in:
lib/krakow/connection_features/snappy_frames.rb

Overview

Snappy-able IO

Constant Summary collapse

IDENTIFIER =

Header identifier

"\x73\x4e\x61\x50\x70\x59".force_encoding('ASCII-8BIT')
IDENTIFIER_SIZE =

Size of identifier

ident_size
CHUNK_TYPE =

Mapping of types

{
  "\xff".force_encoding('ASCII-8BIT') => :identifier,
  "\x00".force_encoding('ASCII-8BIT') => :compressed,
  "\x01".force_encoding('ASCII-8BIT') => :uncompressed
}

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(io, args = {}) ⇒ Io

Create new snappy-able IO

Parameters:

  • io (IO) —

    IO to wrap



39
40
41
42
43
 
# File 'lib/krakow/connection_features/snappy_frames.rb', line 39

def initialize(io, args={})
  @io = io
  @snappy_write_ident = false
  @buffer = ''
end

Dynamic Method Handling

This class handles dynamic methods through the method_missing method

#method_missing(*args) ⇒ Object

Proxy to underlying socket

Parameters:

  • args (Object)

Returns:

  • (Object) 


49
50
51
 
# File 'lib/krakow/connection_features/snappy_frames.rb', line 49

def method_missing(*args)
  io.__send__(*args)
end

Instance Attribute Details

#buffer ⇒ Object (readonly)

Returns the value of attribute buffer.



33
34
35
 
# File 'lib/krakow/connection_features/snappy_frames.rb', line 33

def buffer
  @buffer
end

#io ⇒ Object (readonly)

Returns the value of attribute io.



33
34
35
 
# File 'lib/krakow/connection_features/snappy_frames.rb', line 33

def io
  @io
end

Instance Method Details

#checksum_mask(checksum) ⇒ String

Mask the checksum

Parameters:

  • checksum (String)

Returns:

  • (String) 


57
58
59
 
# File 'lib/krakow/connection_features/snappy_frames.rb', line 57

def checksum_mask(checksum)
  (((checksum >> 15) | (checksum << 17)) + 0xa282ead8) & 0xffffffff
end

#read_stream ⇒ String

Read contents from stream

Returns:

  • (String) 


75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
 
# File 'lib/krakow/connection_features/snappy_frames.rb', line 75

def read_stream
  header = io.recv(4)
  ident = CHUNK_TYPE[header.slice!(0)]
  size = (header << CHUNK_TYPE.key(:compressed)).unpack('L<').first
  content = io.recv(size)
  case ident
  when :identifier
    unless(content == IDENTIFIER)
      raise "Invalid stream identification encountered (content: #{content.inspect})"
    end
    read_stream
  when :compressed
    checksum = content.slice!(0, 4).unpack('L<').first
    deflated = Snappy.inflate(content)
    digest = Digest::CRC32c.new
    digest << deflated
    unless(checksum == checksum_mask(digest.checksum))
      raise 'Checksum mismatch!'
    end
    buffer << deflated
  when :uncompressed
    buffer << content
  end
end

#recv(n) ⇒ String Also known as: read

Receive bytes from the IO

Parameters:

  • n (Integer) —

    nuber of bytes

Returns:

  • (String) 


65
66
67
68
69
 
# File 'lib/krakow/connection_features/snappy_frames.rb', line 65

def recv(n)
  read_stream unless buffer.size >= n
  result = buffer.slice!(0,n)
  result.empty? ? nil : result
end

#send_snappy_identifier ⇒ Integer

Send the identifier for snappy content

Returns:

  • (Integer) —

    bytes written



122
123
124
 
# File 'lib/krakow/connection_features/snappy_frames.rb', line 122

def send_snappy_identifier
  io.write [CHUNK_TYPE.key(:identifier), IDENTIFIER_SIZE, IDENTIFIER].pack('a*a*a*')
end

#write(string) ⇒ Integer

Write string to IO

Parameters:

  • string (String)

Returns:

  • (Integer) —

    number of bytes written



104
105
106
107
108
109
110
111
112
113
114
115
116
117
 
# File 'lib/krakow/connection_features/snappy_frames.rb', line 104

def write(string)
  unless(@snappy_writer_ident)
    send_snappy_identifier
  end
  digest = Digest::CRC32c.new
  digest << string
  content = Snappy.deflate(string)
  size = content.length + 4
  size = [size].pack('L<')
  size.slice!(-1,1)
  checksum = [checksum_mask(digest.checksum)].pack('L<')
  output = [CHUNK_TYPE.key(:compressed), size, checksum, content].pack('a*a*a*a*')
  io.write output
end