Class: CelluloidPubsub::WebServer

Inherits:
Reel::Server::HTTP
  • Object
show all
Includes:
BaseActor
Defined in:
lib/celluloid_pubsub/web_server.rb

Overview

webserver to which socket connects should connect to . the server will dispatch each request into a new Reactor which will handle the action based on the message

Constant Summary collapse

HOST =

The hostname on which the webserver runs on by default

'0.0.0.0'
PATH =

The request path that the webserver accepts by default

'/ws'
CLASSIC_ADAPTER =

The name of the default adapter

'classic'

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Methods included from BaseActor

celluloid_logger_class, celluloid_version, included, setup_actor_supervision, version_less_than_seventeen?

Methods included from Helper

action_subscribe?, fetch_gem_version, filtered_error?, find_loaded_gem, find_loaded_gem_property, get_parsed_version, log_debug, parse_options, setup_celluloid_exception_handler, setup_celluloid_logger, setup_log_file, #succesfull_subscription?, verify_gem_version

Constructor Details

#initialize(options = {}) ⇒ void

receives a list of options that are used to configure the webserver

:nocov:

Parameters:

  • options (Hash) (defaults to: {}) —

    the options that can be used to connect to webser and send additional data

Options Hash (options):

  • :hostname (String) —

    The hostname on which the webserver runs on

  • :port (Integer) —

    The port on which the webserver runs on

  • :path (String) —

    The request path that the webserver accepts

  • :spy (Boolean) —

    Enable this only if you want to enable debugging for the webserver



41
42
43
44
45
46
47
48
49
 
# File 'lib/celluloid_pubsub/web_server.rb', line 41

def initialize(options = {})
  Celluloid.boot unless Celluloid.running?
  @server_options = parse_options(options)
  @subscribers = {}
  @mutex = Mutex.new
  setup_celluloid_logger
  debug "CelluloidPubsub::WebServer example starting on #{hostname}:#{port}"
  super(hostname, port, { spy: spy, backlog: backlog }, &method(:on_connection))
end

Instance Attribute Details

#mutex ⇒ Object

Returns the value of attribute mutex.



26
27
28
 
# File 'lib/celluloid_pubsub/web_server.rb', line 26

def mutex
  @mutex
end

#options ⇒ Hash

Returns options used to configure the webserver.

Returns:

  • (Hash) —

    options used to configure the webserver



16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
 
# File 'lib/celluloid_pubsub/web_server.rb', line 16

class WebServer < Reel::Server::HTTP
  include CelluloidPubsub::BaseActor

  # The hostname on which the webserver runs on by default
  HOST = '0.0.0.0'
  # The request path that the webserver accepts by default
  PATH = '/ws'
    # The name of the default adapter
  CLASSIC_ADAPTER = 'classic'

  attr_accessor :server_options, :subscribers, :mutex
  finalizer :shutdown
  #  receives a list of options that are used to configure the webserver
  #
  # @param  [Hash]  options the options that can be used to connect to webser and send additional data
  # @option options [String]:hostname The hostname on which the webserver runs on
  # @option options [Integer] :port The port on which the webserver runs on
  # @option options [String] :path The request path that the webserver accepts
  # @option options [Boolean] :spy Enable this only if you want to enable debugging for the webserver
  #
  # @return [void]
  #
  # @api public
  #
  # :nocov:
  def initialize(options = {})
    Celluloid.boot unless Celluloid.running?
    @server_options = parse_options(options)
    @subscribers = {}
    @mutex = Mutex.new
    setup_celluloid_logger
    debug "CelluloidPubsub::WebServer example starting on #{hostname}:#{port}"
    super(hostname, port, { spy: spy, backlog: backlog }, &method(:on_connection))
  end

  # the method will  return the socket conection opened on the unused port
  #
  #
  # @return [TCPServer]  return the socket connection opened on a random port
  #
  # @api public
  def self.open_socket_on_unused_port
    infos = ::Socket::getaddrinfo("localhost", nil, Socket::AF_UNSPEC, Socket::SOCK_STREAM, 0, Socket::AI_PASSIVE)
    families = Hash[*infos.collect { |af, *_| af }.uniq.zip([]).flatten]

    return ::TCPServer.open('0.0.0.0', 0) if families.has_key?('AF_INET')
    return ::TCPServer.open('::', 0) if families.has_key?('AF_INET6')
    return ::TCPServer.open(0)
  end

  # the method get from the socket connection that is already opened the port used.
  # @see #open_socket_on_unused_port
  #
  # @return [Integer]  returns the port that can be used to issue new connection
  #
  # @api public
  def self.find_unused_port
    @@unused_port ||= begin
      socket = open_socket_on_unused_port
      port = socket.addr[1]
      socket.close
      port
    end
  end

  # this method is overriden from the Reel::Server::HTTP in order to set the spy to the celluloid logger
  # before the connection is accepted.
  # @see #handle_connection
  # @api public
  def run
    @spy = Celluloid.logger if spy
    loop { async.handle_connection @server.accept }
  end

  # the method will  return true if redis can be used otherwise false
  #
  #
  # @return [Boolean]  return true if redis can be used otherwise false
  #
  # @api public
  def adapter
    @adapter ||= @server_options.fetch('adapter', CelluloidPubsub::WebServer::CLASSIC_ADAPTER)
    @adapter.present? ? @adapter : CelluloidPubsub::WebServer::CLASSIC_ADAPTER
  end

  # the method will return true if debug is enabled otherwise false
  #
  #
  # @return [Boolean] returns true if debug is enabled otherwise false
  #
  # @api public
  def debug_enabled?
    @debug_enabled = @server_options.fetch('enable_debug', false)
    @debug_enabled == true
  end

  # the method will terminate the current actor
  #
  #
  # @return [void]
  #
  # @api public
  def shutdown
    debug "#{self.class} tries to 'shudown'"
    terminate
  end

  # the method will return the file path of the log file where debug messages will be printed
  #
  #
  # @return [String] returns the file path of the log file where debug messages will be printed
  #
  # @api public
  def log_file_path
    @log_file_path = @server_options.fetch('log_file_path', nil)
  end

  # the method will return the hostname on which the server is running on
  #
  #
  # @return [String] returns the hostname on which the server is running on
  #
  # @api public
  def hostname
    @hostname = @server_options.fetch('hostname', CelluloidPubsub::WebServer::HOST)
  end

  # the method will return the port on which will accept connections
  #
  #
  # @return [String] returns the port on which will accept connections
  #
  # @api public
  def port
    @port ||= @server_options.fetch('port', nil) || self.class.find_unused_port
  end

  # the method will return the URL path on which will acceept connections
  #
  #
  # @return [String] returns the URL path on which will acceept connections
  #
  # @api public
  def path
    @path = @server_options.fetch('path', CelluloidPubsub::WebServer::PATH)
  end

  # the method will return true if connection to the server should be spied upon
  #
  #
  # @return [Boolean] returns true if connection to the server should be spied upon, otherwise false
  #
  # @api public
  def spy
    @spy = @server_options.fetch('spy', false)
  end

  # the method will return the number of connections allowed to the server
  #
  #
  # @return [Integer] returns the number of connections allowed to the server
  #
  # @api public
  def backlog
    @backlog = @server_options.fetch('backlog', 1024)
  end


  #  callback that will execute when receiving new conections
  # If the connections is a websocket will call method {#route_websocket}
  # and if the connection is HTTP will call method {#route_request}
  # For websocket connections , the connection is detached from the server and dispatched to another actor
  #
  # @see #route_websocket
  # @see #route_request
  #
  # @param [Reel::WebSocket] connection The connection that was made to the webserver
  #
  # @return [void]
  #
  # @api public
  def on_connection(connection)
    while request = connection.request
      if request.websocket?
        log_debug "#{self.class} Received a WebSocket connection"

        # We're going to hand off this connection to another actor (Writer/Reader)
        # However, initially Reel::Connections are "attached" to the
        # Reel::Server::HTTP actor, meaning that the server manages the connection
        # lifecycle (e.g. error handling) for us.
        #
        # If we want to hand this connection off to another actor, we first
        # need to detach it from the Reel::Server (in this case, Reel::Server::HTTP)
        connection.detach
        dispatch_websocket_request(request)
        return
      else
        route_request connection, request
      end
    end
  end

  #  returns the reactor class that will handle the connection depending if redis is enabled or not
  # @see #redis_enabled?
  #
  # @return [Class]  returns the reactor class that will handle the connection depending if redis is enabled or not
  #
  # @api public
  def reactor_class
    adapter == CelluloidPubsub::WebServer::CLASSIC_ADAPTER ? CelluloidPubsub::Reactor : "CelluloidPubsub::#{adapter.camelize}Reactor".constantize
  end

  # method will instantiate a new reactor object, will link the reactor to the current actor and will dispatch the request to the reactor
  # @see #route_websocket
  #
  # @param [Reel::WebSocket] request The request that was made to the webserver
  #
  # @return [void]
  #
  # @api public
  def dispatch_websocket_request(request)
    reactor = reactor_class.new
    Actor.current.link reactor
    route_websocket(reactor, request.websocket)
  end

  #  HTTP connections are not accepted so this method will show 404 message "Not Found"
  #
  # @param [Reel::WebSocket] connection The HTTP connection that was received
  # @param [Reel::Request] request The request that was made to the webserver and contains the type , the url, and the parameters
  #
  # @return [void]
  #
  # @api public
  def route_request(connection, request)
    log_debug "404 Not Found: #{request.path}"
    connection.respond :not_found, 'Not found'
  end

  #  If the socket url matches with the one accepted by the server, it will dispatch the socket connection to a new reactor {CelluloidPubsub::Reactor#work}
  # The new actor is linked to the webserver
  #
  # @see CelluloidPubsub::Reactor#work
  #
  #  @param [Reel::WebSocket] socket The  web socket connection that was received
  #
  # @return [void]
  #
  # @api public
  def route_websocket(reactor, socket)
     url = socket.url
     if url == path || url == "/?"
      reactor.async.work(socket, Actor.current)
     else
       log_debug "Received invalid WebSocket request for: #{url}"
       socket.close
     end
  end

  # If the message can be parsed into a Hash it will respond to the reactor's websocket connection with the same message in JSON format
  # otherwise will try send the message how it is and escaped into JSON format
  #
  # @param [CelluloidPubsub::Reactor] reactor The  reactor that received an unhandled message
  # @param [Object] data The message that the reactor could not handle
  #
  # @return [void]
  #
  # @api public
  def handle_dispatched_message(reactor, data)
    log_debug "#{self.class} trying to dispatch message  #{data.inspect}"
    message = reactor.parse_json_data(data)
    if message.present? && message.is_a?(Hash)
      final_data = message.to_json
    else
      final_data = data.to_json
    end
    reactor.websocket << final_data
  end
end

#server_options ⇒ Object

Returns the value of attribute server_options.



26
27
28
 
# File 'lib/celluloid_pubsub/web_server.rb', line 26

def server_options
  @server_options
end

#subscribers ⇒ Hash

Returns The hostname on which the webserver runs on.

Returns:

  • (Hash) —

    The hostname on which the webserver runs on



16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
 
# File 'lib/celluloid_pubsub/web_server.rb', line 16

class WebServer < Reel::Server::HTTP
  include CelluloidPubsub::BaseActor

  # The hostname on which the webserver runs on by default
  HOST = '0.0.0.0'
  # The request path that the webserver accepts by default
  PATH = '/ws'
    # The name of the default adapter
  CLASSIC_ADAPTER = 'classic'

  attr_accessor :server_options, :subscribers, :mutex
  finalizer :shutdown
  #  receives a list of options that are used to configure the webserver
  #
  # @param  [Hash]  options the options that can be used to connect to webser and send additional data
  # @option options [String]:hostname The hostname on which the webserver runs on
  # @option options [Integer] :port The port on which the webserver runs on
  # @option options [String] :path The request path that the webserver accepts
  # @option options [Boolean] :spy Enable this only if you want to enable debugging for the webserver
  #
  # @return [void]
  #
  # @api public
  #
  # :nocov:
  def initialize(options = {})
    Celluloid.boot unless Celluloid.running?
    @server_options = parse_options(options)
    @subscribers = {}
    @mutex = Mutex.new
    setup_celluloid_logger
    debug "CelluloidPubsub::WebServer example starting on #{hostname}:#{port}"
    super(hostname, port, { spy: spy, backlog: backlog }, &method(:on_connection))
  end

  # the method will  return the socket conection opened on the unused port
  #
  #
  # @return [TCPServer]  return the socket connection opened on a random port
  #
  # @api public
  def self.open_socket_on_unused_port
    infos = ::Socket::getaddrinfo("localhost", nil, Socket::AF_UNSPEC, Socket::SOCK_STREAM, 0, Socket::AI_PASSIVE)
    families = Hash[*infos.collect { |af, *_| af }.uniq.zip([]).flatten]

    return ::TCPServer.open('0.0.0.0', 0) if families.has_key?('AF_INET')
    return ::TCPServer.open('::', 0) if families.has_key?('AF_INET6')
    return ::TCPServer.open(0)
  end

  # the method get from the socket connection that is already opened the port used.
  # @see #open_socket_on_unused_port
  #
  # @return [Integer]  returns the port that can be used to issue new connection
  #
  # @api public
  def self.find_unused_port
    @@unused_port ||= begin
      socket = open_socket_on_unused_port
      port = socket.addr[1]
      socket.close
      port
    end
  end

  # this method is overriden from the Reel::Server::HTTP in order to set the spy to the celluloid logger
  # before the connection is accepted.
  # @see #handle_connection
  # @api public
  def run
    @spy = Celluloid.logger if spy
    loop { async.handle_connection @server.accept }
  end

  # the method will  return true if redis can be used otherwise false
  #
  #
  # @return [Boolean]  return true if redis can be used otherwise false
  #
  # @api public
  def adapter
    @adapter ||= @server_options.fetch('adapter', CelluloidPubsub::WebServer::CLASSIC_ADAPTER)
    @adapter.present? ? @adapter : CelluloidPubsub::WebServer::CLASSIC_ADAPTER
  end

  # the method will return true if debug is enabled otherwise false
  #
  #
  # @return [Boolean] returns true if debug is enabled otherwise false
  #
  # @api public
  def debug_enabled?
    @debug_enabled = @server_options.fetch('enable_debug', false)
    @debug_enabled == true
  end

  # the method will terminate the current actor
  #
  #
  # @return [void]
  #
  # @api public
  def shutdown
    debug "#{self.class} tries to 'shudown'"
    terminate
  end

  # the method will return the file path of the log file where debug messages will be printed
  #
  #
  # @return [String] returns the file path of the log file where debug messages will be printed
  #
  # @api public
  def log_file_path
    @log_file_path = @server_options.fetch('log_file_path', nil)
  end

  # the method will return the hostname on which the server is running on
  #
  #
  # @return [String] returns the hostname on which the server is running on
  #
  # @api public
  def hostname
    @hostname = @server_options.fetch('hostname', CelluloidPubsub::WebServer::HOST)
  end

  # the method will return the port on which will accept connections
  #
  #
  # @return [String] returns the port on which will accept connections
  #
  # @api public
  def port
    @port ||= @server_options.fetch('port', nil) || self.class.find_unused_port
  end

  # the method will return the URL path on which will acceept connections
  #
  #
  # @return [String] returns the URL path on which will acceept connections
  #
  # @api public
  def path
    @path = @server_options.fetch('path', CelluloidPubsub::WebServer::PATH)
  end

  # the method will return true if connection to the server should be spied upon
  #
  #
  # @return [Boolean] returns true if connection to the server should be spied upon, otherwise false
  #
  # @api public
  def spy
    @spy = @server_options.fetch('spy', false)
  end

  # the method will return the number of connections allowed to the server
  #
  #
  # @return [Integer] returns the number of connections allowed to the server
  #
  # @api public
  def backlog
    @backlog = @server_options.fetch('backlog', 1024)
  end


  #  callback that will execute when receiving new conections
  # If the connections is a websocket will call method {#route_websocket}
  # and if the connection is HTTP will call method {#route_request}
  # For websocket connections , the connection is detached from the server and dispatched to another actor
  #
  # @see #route_websocket
  # @see #route_request
  #
  # @param [Reel::WebSocket] connection The connection that was made to the webserver
  #
  # @return [void]
  #
  # @api public
  def on_connection(connection)
    while request = connection.request
      if request.websocket?
        log_debug "#{self.class} Received a WebSocket connection"

        # We're going to hand off this connection to another actor (Writer/Reader)
        # However, initially Reel::Connections are "attached" to the
        # Reel::Server::HTTP actor, meaning that the server manages the connection
        # lifecycle (e.g. error handling) for us.
        #
        # If we want to hand this connection off to another actor, we first
        # need to detach it from the Reel::Server (in this case, Reel::Server::HTTP)
        connection.detach
        dispatch_websocket_request(request)
        return
      else
        route_request connection, request
      end
    end
  end

  #  returns the reactor class that will handle the connection depending if redis is enabled or not
  # @see #redis_enabled?
  #
  # @return [Class]  returns the reactor class that will handle the connection depending if redis is enabled or not
  #
  # @api public
  def reactor_class
    adapter == CelluloidPubsub::WebServer::CLASSIC_ADAPTER ? CelluloidPubsub::Reactor : "CelluloidPubsub::#{adapter.camelize}Reactor".constantize
  end

  # method will instantiate a new reactor object, will link the reactor to the current actor and will dispatch the request to the reactor
  # @see #route_websocket
  #
  # @param [Reel::WebSocket] request The request that was made to the webserver
  #
  # @return [void]
  #
  # @api public
  def dispatch_websocket_request(request)
    reactor = reactor_class.new
    Actor.current.link reactor
    route_websocket(reactor, request.websocket)
  end

  #  HTTP connections are not accepted so this method will show 404 message "Not Found"
  #
  # @param [Reel::WebSocket] connection The HTTP connection that was received
  # @param [Reel::Request] request The request that was made to the webserver and contains the type , the url, and the parameters
  #
  # @return [void]
  #
  # @api public
  def route_request(connection, request)
    log_debug "404 Not Found: #{request.path}"
    connection.respond :not_found, 'Not found'
  end

  #  If the socket url matches with the one accepted by the server, it will dispatch the socket connection to a new reactor {CelluloidPubsub::Reactor#work}
  # The new actor is linked to the webserver
  #
  # @see CelluloidPubsub::Reactor#work
  #
  #  @param [Reel::WebSocket] socket The  web socket connection that was received
  #
  # @return [void]
  #
  # @api public
  def route_websocket(reactor, socket)
     url = socket.url
     if url == path || url == "/?"
      reactor.async.work(socket, Actor.current)
     else
       log_debug "Received invalid WebSocket request for: #{url}"
       socket.close
     end
  end

  # If the message can be parsed into a Hash it will respond to the reactor's websocket connection with the same message in JSON format
  # otherwise will try send the message how it is and escaped into JSON format
  #
  # @param [CelluloidPubsub::Reactor] reactor The  reactor that received an unhandled message
  # @param [Object] data The message that the reactor could not handle
  #
  # @return [void]
  #
  # @api public
  def handle_dispatched_message(reactor, data)
    log_debug "#{self.class} trying to dispatch message  #{data.inspect}"
    message = reactor.parse_json_data(data)
    if message.present? && message.is_a?(Hash)
      final_data = message.to_json
    else
      final_data = data.to_json
    end
    reactor.websocket << final_data
  end
end

Class Method Details

.find_unused_port ⇒ Integer

the method get from the socket connection that is already opened the port used.

Returns:

  • (Integer) —

    returns the port that can be used to issue new connection

See Also:

  • #open_socket_on_unused_port


72
73
74
75
76
77
78
79
 
# File 'lib/celluloid_pubsub/web_server.rb', line 72

def self.find_unused_port
  @@unused_port ||= begin
    socket = open_socket_on_unused_port
    port = socket.addr[1]
    socket.close
    port
  end
end

.open_socket_on_unused_port ⇒ TCPServer

the method will return the socket conection opened on the unused port

Returns:

  • (TCPServer) —

    return the socket connection opened on a random port



57
58
59
60
61
62
63
64
 
# File 'lib/celluloid_pubsub/web_server.rb', line 57

def self.open_socket_on_unused_port
  infos = ::Socket::getaddrinfo("localhost", nil, Socket::AF_UNSPEC, Socket::SOCK_STREAM, 0, Socket::AI_PASSIVE)
  families = Hash[*infos.collect { |af, *_| af }.uniq.zip([]).flatten]

  return ::TCPServer.open('0.0.0.0', 0) if families.has_key?('AF_INET')
  return ::TCPServer.open('::', 0) if families.has_key?('AF_INET6')
  return ::TCPServer.open(0)
end

Instance Method Details

#adapter ⇒ Boolean

the method will return true if redis can be used otherwise false

Returns:

  • (Boolean) —

    return true if redis can be used otherwise false



96
97
98
99
 
# File 'lib/celluloid_pubsub/web_server.rb', line 96

def adapter
  @adapter ||= @server_options.fetch('adapter', CelluloidPubsub::WebServer::CLASSIC_ADAPTER)
  @adapter.present? ? @adapter : CelluloidPubsub::WebServer::CLASSIC_ADAPTER
end

#backlog ⇒ Integer

the method will return the number of connections allowed to the server

Returns:

  • (Integer) —

    returns the number of connections allowed to the server



179
180
181
 
# File 'lib/celluloid_pubsub/web_server.rb', line 179

def backlog
  @backlog = @server_options.fetch('backlog', 1024)
end

#debug_enabled? ⇒ Boolean

the method will return true if debug is enabled otherwise false

Returns:

  • (Boolean) —

    returns true if debug is enabled otherwise false



107
108
109
110
 
# File 'lib/celluloid_pubsub/web_server.rb', line 107

def debug_enabled?
  @debug_enabled = @server_options.fetch('enable_debug', false)
  @debug_enabled == true
end

#dispatch_websocket_request(request) ⇒ void

This method returns an undefined value.

method will instantiate a new reactor object, will link the reactor to the current actor and will dispatch the request to the reactor

Parameters:

  • request (Reel::WebSocket) —

    The request that was made to the webserver

See Also:



236
237
238
239
240
 
# File 'lib/celluloid_pubsub/web_server.rb', line 236

def dispatch_websocket_request(request)
  reactor = reactor_class.new
  Actor.current.link reactor
  route_websocket(reactor, request.websocket)
end

#handle_dispatched_message(reactor, data) ⇒ void

This method returns an undefined value.

If the message can be parsed into a Hash it will respond to the reactor’s websocket connection with the same message in JSON format otherwise will try send the message how it is and escaped into JSON format

Parameters:

  • reactor (CelluloidPubsub::Reactor) —

    The reactor that received an unhandled message

  • data (Object) —

    The message that the reactor could not handle



284
285
286
287
288
289
290
291
292
293
 
# File 'lib/celluloid_pubsub/web_server.rb', line 284

def handle_dispatched_message(reactor, data)
  log_debug "#{self.class} trying to dispatch message  #{data.inspect}"
  message = reactor.parse_json_data(data)
  if message.present? && message.is_a?(Hash)
    final_data = message.to_json
  else
    final_data = data.to_json
  end
  reactor.websocket << final_data
end

#hostname ⇒ String

the method will return the hostname on which the server is running on

Returns:

  • (String) —

    returns the hostname on which the server is running on



139
140
141
 
# File 'lib/celluloid_pubsub/web_server.rb', line 139

def hostname
  @hostname = @server_options.fetch('hostname', CelluloidPubsub::WebServer::HOST)
end

#log_file_path ⇒ String

the method will return the file path of the log file where debug messages will be printed

Returns:

  • (String) —

    returns the file path of the log file where debug messages will be printed



129
130
131
 
# File 'lib/celluloid_pubsub/web_server.rb', line 129

def log_file_path
  @log_file_path = @server_options.fetch('log_file_path', nil)
end

#on_connection(connection) ⇒ void

This method returns an undefined value.

callback that will execute when receiving new conections If the connections is a websocket will call method #route_websocket and if the connection is HTTP will call method #route_request For websocket connections , the connection is detached from the server and dispatched to another actor

Parameters:

  • connection (Reel::WebSocket) —

    The connection that was made to the webserver

See Also:



197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
 
# File 'lib/celluloid_pubsub/web_server.rb', line 197

def on_connection(connection)
  while request = connection.request
    if request.websocket?
      log_debug "#{self.class} Received a WebSocket connection"

      # We're going to hand off this connection to another actor (Writer/Reader)
      # However, initially Reel::Connections are "attached" to the
      # Reel::Server::HTTP actor, meaning that the server manages the connection
      # lifecycle (e.g. error handling) for us.
      #
      # If we want to hand this connection off to another actor, we first
      # need to detach it from the Reel::Server (in this case, Reel::Server::HTTP)
      connection.detach
      dispatch_websocket_request(request)
      return
    else
      route_request connection, request
    end
  end
end

#path ⇒ String

the method will return the URL path on which will acceept connections

Returns:

  • (String) —

    returns the URL path on which will acceept connections



159
160
161
 
# File 'lib/celluloid_pubsub/web_server.rb', line 159

def path
  @path = @server_options.fetch('path', CelluloidPubsub::WebServer::PATH)
end

#port ⇒ String

the method will return the port on which will accept connections

Returns:

  • (String) —

    returns the port on which will accept connections



149
150
151
 
# File 'lib/celluloid_pubsub/web_server.rb', line 149

def port
  @port ||= @server_options.fetch('port', nil) || self.class.find_unused_port
end

#reactor_class ⇒ Class

returns the reactor class that will handle the connection depending if redis is enabled or not

Returns:

  • (Class) —

    returns the reactor class that will handle the connection depending if redis is enabled or not

See Also:

  • #redis_enabled?


224
225
226
 
# File 'lib/celluloid_pubsub/web_server.rb', line 224

def reactor_class
  adapter == CelluloidPubsub::WebServer::CLASSIC_ADAPTER ? CelluloidPubsub::Reactor : "CelluloidPubsub::#{adapter.camelize}Reactor".constantize
end

#route_request(connection, request) ⇒ void

This method returns an undefined value.

HTTP connections are not accepted so this method will show 404 message “Not Found”

Parameters:

  • connection (Reel::WebSocket) —

    The HTTP connection that was received

  • request (Reel::Request) —

    The request that was made to the webserver and contains the type , the url, and the parameters



250
251
252
253
 
# File 'lib/celluloid_pubsub/web_server.rb', line 250

def route_request(connection, request)
  log_debug "404 Not Found: #{request.path}"
  connection.respond :not_found, 'Not found'
end

#route_websocket(reactor, socket) ⇒ void

This method returns an undefined value.

If the socket url matches with the one accepted by the server, it will dispatch the socket connection to a new reactor Reactor#work The new actor is linked to the webserver

See Also:



265
266
267
268
269
270
271
272
273
 
# File 'lib/celluloid_pubsub/web_server.rb', line 265

def route_websocket(reactor, socket)
   url = socket.url
   if url == path || url == "/?"
    reactor.async.work(socket, Actor.current)
   else
     log_debug "Received invalid WebSocket request for: #{url}"
     socket.close
   end
end

#run ⇒ Object

this method is overriden from the Reel::Server::HTTP in order to set the spy to the celluloid logger before the connection is accepted.

See Also:

  • #handle_connection


85
86
87
88
 
# File 'lib/celluloid_pubsub/web_server.rb', line 85

def run
  @spy = Celluloid.logger if spy
  loop { async.handle_connection @server.accept }
end

#shutdown ⇒ void

This method returns an undefined value.

the method will terminate the current actor



118
119
120
121
 
# File 'lib/celluloid_pubsub/web_server.rb', line 118

def shutdown
  debug "#{self.class} tries to 'shudown'"
  terminate
end

#spy ⇒ Boolean

the method will return true if connection to the server should be spied upon

Returns:

  • (Boolean) —

    returns true if connection to the server should be spied upon, otherwise false



169
170
171
 
# File 'lib/celluloid_pubsub/web_server.rb', line 169

def spy
  @spy = @server_options.fetch('spy', false)
end