class WebSocketRb::Protocol::FramesHandler

Public Class Methods

new(connections, connection, routes, frames_sender, sandbox) click to toggle source
# File lib/web_socket_rb/protocol/frames_handler.rb, line 12
def initialize(connections, connection, routes, frames_sender, sandbox)
  @connections      = connections
  @conn             = connection
  @routes           = routes
  @frames_sender    = frames_sender
  @sandbox          = sandbox
  @threads          = []
  @pending_pings    = []
  @mutex            = Mutex.new
end

Public Instance Methods

run() click to toggle source

Run reading and sending frames

# File lib/web_socket_rb/protocol/frames_handler.rb, line 24
def run
  @threads << Thread.new { verify_status }
  @threads << Thread.new { process_incoming_frames }
  @threads.each(&:join)
end

Private Instance Methods

process_incoming_frame(frame) click to toggle source

Process incoming frame

# File lib/web_socket_rb/protocol/frames_handler.rb, line 52
def process_incoming_frame(frame)
  if frame.ping?

    # Send PONG frame if requested PING frame
    App.logger.info('Frames handler') { 'Replying for PING request' }
    pong_frame = Service::BuildPongFrameService.new(frame).run
    @frames_sender.frame_to_send(pong_frame)

  elsif frame.close?

    # Send CLOSE frame if requested CLOSE frame
    App.logger.info('Frames handler') { 'Replying for CLOSE request' }
    close_frame = Service::BuildCloseFrameService.new(frame).run
    @frames_sender.frame_to_send(close_frame)
    @threads.each(&:exit)

  elsif frame.pong?

    # Verify pending PINGs and find one with the same payload data
    App.logger.info('Frames handler') {'Verifying PINGS'}
    @mutex.synchronize do
      pending_ping = @pending_pings.reject! do |ping_frame|
        frame.payload_data == ping_frame.payload_data
      end
      @threads.each(&:exit) if pending_ping.nil?
    end

  elsif frame.text? || frame.binary?

    # Call subscribe block
    App.logger.info('Frames handler') { 'Processing frame' }
    subscribe_block = @routes.subscribes[frame.destination]
    App.logger.info('Frames handler') { frame.message }
    Thread.new { @sandbox.instance_exec(frame.message, &subscribe_block) }

  end
end
process_incoming_frames() click to toggle source

Read incoming frames

# File lib/web_socket_rb/protocol/frames_handler.rb, line 44
def process_incoming_frames
  loop do
    frame = Service::ReadFrameService.new(@conn).run
    process_incoming_frame(frame) unless frame.nil?
  end
end
verify_status() click to toggle source

Send PING frame every 10 seconds to verify connection

# File lib/web_socket_rb/protocol/frames_handler.rb, line 33
def verify_status
  loop do
    frame = Service::BuildPingFrameService.new.run
    @frames_sender.frame_to_send(frame)
    @mutex.synchronize { @pending_pings << frame }
    sleep(10)
    @mutex.synchronize { @threads.reverse.each(&:kill) if @pending_pings.include?(frame) }
  end
end