class PipeRpc::Hub::Socket

Public Class Methods

new(hub, args = {}) click to toggle source
# File lib/pipe_rpc/hub_socket.rb, line 3
def initialize(hub, args = {})
  @hub = hub
  @input = args.fetch(:input)
  @output = args.fetch(:output)
  @on_sent = []
  @on_received = []
  @closed_reason = "lost connection to other side"
end

Public Instance Methods

close(reason) click to toggle source
# File lib/pipe_rpc/hub_socket.rb, line 54
def close(reason)
  @closed_reason = reason
  @input.close
  @output.close
end
off_received(callback) click to toggle source
# File lib/pipe_rpc/hub_socket.rb, line 50
def off_received(callback)
  @on_received.delete(callback)
end
off_sent(callback) click to toggle source
# File lib/pipe_rpc/hub_socket.rb, line 41
def off_sent(callback)
  @on_sent.delete(callback)
end
on_received(&on_received) click to toggle source
# File lib/pipe_rpc/hub_socket.rb, line 45
def on_received(&on_received)
  @on_received << on_received
  on_received
end
on_sent(&on_sent) click to toggle source
# File lib/pipe_rpc/hub_socket.rb, line 36
def on_sent(&on_sent)
  @on_sent << on_sent
  on_sent
end
read() click to toggle source
# File lib/pipe_rpc/hub_socket.rb, line 12
def read
  raise ClosedError.new(@closed_reason) if @input.closed?

  # infinite #readpartial that also works with mruby-restricted_io
  packed = ''
  loop do
    packed << @input.sysread(2*16)
    break unless @input.class.select([@input], nil, nil, 0)
  end

  unpack(packed).tap do |unpacked|
    @on_received.each{ |cb| cb.call(unpacked) }
  end
rescue MessagePack::Error => e
  write ErrorResponse.new(id: nil, error: { code: -32700, data: { message: "MessagePack: #{e.message}" } })
end
unpack(packed) click to toggle source
# File lib/pipe_rpc/hub_socket.rb, line 60
def unpack(packed)
  if ::Object.const_defined?(:MRUBY_VERSION)
    MessagePack.unpack(packed, true) # to create unknown symbols in mruby
  else
    MessagePack.unpack(packed)
  end
end
write(obj) click to toggle source
# File lib/pipe_rpc/hub_socket.rb, line 29
def write(obj)
  raise ClosedError.new(@closed_reason) if @output.closed?
  payload = obj.to_h
  @on_sent.each{ |callback| callback.call(payload) }
  @output.syswrite payload.to_msgpack
end