Class: Listen::TCP::Broadcaster

Inherits:
Object
  • Object
show all
Includes:
Celluloid::IO
Defined in:
lib/listen/tcp/broadcaster.rb

Instance Method Summary collapse

Constructor Details

#initialize(host, port) ⇒ Broadcaster

Initializes a Celluloid::IO-powered TCP-broadcaster

Note: Listens on all addresses when host is nil

Parameters:

  • host (String) —

    to broadcast on

  • port (String) —

    to broadcast on



17
18
19
20
21
22
23
24
25
26
 
# File 'lib/listen/tcp/broadcaster.rb', line 17

def initialize(host, port)
  @sockets = []
  _log :debug, format('Broadcaster: tcp server listening on: %s:%s',
                      host, port)
  @server = TCPServer.new(host, port)
rescue
  _log :error, format('Broadcaster.initialize: %s:%s', $ERROR_INFO,
                      $ERROR_POSITION * "\n")
  raise
end

Instance Method Details

#_log(type, message) ⇒ Object (private) 



66
67
68
 
# File 'lib/listen/tcp/broadcaster.rb', line 66

def _log(type, message)
  Celluloid::Logger.send(type, message)
end

#_unicast(socket, payload) ⇒ Object (private) 



70
71
72
73
74
75
76
 
# File 'lib/listen/tcp/broadcaster.rb', line 70

def _unicast(socket, payload)
  socket.write(payload)
  true
rescue IOError, Errno::ECONNRESET, Errno::EPIPE
  _log :debug, "Broadcaster failed: #{socket.inspect}"
  false
end

#broadcast(payload) ⇒ Object

Broadcasts given payload to all connected sockets



44
45
46
47
48
49
 
# File 'lib/listen/tcp/broadcaster.rb', line 44

def broadcast(payload)
  active_sockets = @sockets.select do |socket|
    _unicast(socket, payload)
  end
  @sockets.replace(active_sockets)
end

#finalize ⇒ Object

Cleans up sockets and server



34
35
36
37
38
39
40
41
 
# File 'lib/listen/tcp/broadcaster.rb', line 34

def finalize
  @sockets.map(&:close) if @sockets
  @sockets = nil

  return unless @server
  @server.close
  @server = nil
end

#run ⇒ Object

Continuously accept and handle incoming connections



52
53
54
55
56
57
58
59
60
61
62
 
# File 'lib/listen/tcp/broadcaster.rb', line 52

def run
  while (socket = @server.accept)
    @sockets << socket
  end
rescue Celluloid::Task::TerminatedError
  _log :debug, "TCP adapter was terminated: #{$ERROR_INFO}"
rescue
  _log :error, format('Broadcaster.run: %s:%s', $ERROR_INFO,
                      $ERROR_POSITION * "\n")
  raise
end

#start ⇒ Object

Asynchronously start accepting connections



29
30
31
 
# File 'lib/listen/tcp/broadcaster.rb', line 29

def start
  async.run
end