beats_empty_writer.rb

# encoding: utf-8

require "socket"
require "thread"
require "zlib"
require "json"

Thread.abort_on_exception = true
HOST="127.0.0.1"
PORT=3333

module Lumberjack
  SEQUENCE_MAX = (2**32-1).freeze

  class Client
    def initialize
      @sequence = 0
      @last_ack = 0
      @socket = connect
    end

    private
    def connect
      socket = TCPSocket.new(HOST, PORT)
    end

    public
    def write(elements)
      elements = [elements] if elements.is_a?(Hash)
      puts "set window to #{elements.size}. sequence #{@sequence}"
      send_window_size(elements.size)

      payload = elements.map { |element| JsonEncoder.to_frame(element, inc) }.join
      send_payload(payload) unless elements.empty?
      ack(elements.size)
    end

    private
    def ack(size)
      _, type = read_version_and_type
      raise "Whoa we shouldn't get this frame: #{type}" if type != "A"
      @last_ack = read_last_ack
    end

    private
    def read_version_and_type
      version = @socket.read(1)
      type    = @socket.read(1)
      [version, type]
    end

    private
    def read_last_ack
      @socket.read(4).unpack("N").first
    end

    private
    def inc
      @sequence = 0 if @sequence + 1 > Lumberjack::SEQUENCE_MAX
      @sequence = @sequence + 1
    end

    private
    def send_window_size(size)
      @socket.syswrite(["2", "W", size].pack("AAN"))
    end

    private
    def send_payload(payload)
      payload_size = payload.size
      written = 0
      while written < payload_size
        written += @socket.syswrite(payload[written..-1])
      end
    end

    public
    def close
      @socket.close
    end
  end

  module JsonEncoder
    def self.to_frame(hash, sequence)
      json = hash.to_json
      json_length = json.bytesize
      pack = "AANNA#{json_length}"
      frame = ["2", "J", sequence, json_length, json]
      frame.pack(pack)
    end
  end

end

client = Lumberjack::Client.new
data = [
  { "a" => 1, "b" => 2, },
  { "c" => 3, "d" => 4, },
]

ack = client.write(data)
puts ack
ack = client.write([])
puts ack
ack = client.write(data)
puts ack
sleep 1
client.close
puts "Done"
sleep 5