Last active

Embed URL

HTTPS clone URL

SSH clone URL

You can clone with HTTPS or SSH.

Download Gist

Reel Websocket Server using PG Listen/Notify for crude pubsub

View reel_ws_pg_example.rb
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 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
require 'rubygems'
require 'bundler/setup'
require 'reel'
require 'celluloid/io'
require 'pg'
 
module PGNotifications
def self.included(actor)
actor.send(:include, Celluloid::IO)
end
 
def actions
@actions ||= {}
end
 
def pg_connection
@pg_connection ||= PG.connect( dbname: 'ws_example' )
end
 
def notify(channel, value)
pg_connection.exec("NOTIFY #{channel}, '#{value}';")
end
 
def listen(channel, action)
actions[channel] = action
pg_connection.exec("LISTEN #{channel}")
end
 
def start_listening
info "Starting Listening"
 
@listening = true
 
wait_for_notify do |channel, pid, payload|
info "Received notification: #{[channel, pid, payload].inspect}"
send(actions[channel], channel, payload)
end
end
 
def stop_listening
@listening = false
end
 
def wait_for_notify(&block)
io = pg_connection.socket_io
 
while @listening do
Celluloid::IO.wait_readable(io) # blocks execution, but unblocks this actor
pg_connection.consume_input # fetch any input on this connection
 
while notification = pg_connection.notifies do
block.call(*[
notification[:relname], # channel
notification[:be_pid], # pid
notification[:extra] # payload
])
end
end
end
 
def unlisten(channel)
(@listening ||= {})[channel] = false
 
pg_connection.exec("UNLISTEN #{channel}")
end
end
 
class TimeServer
include Celluloid
include PGNotifications
include Celluloid::Logger
 
def initialize
async.run
end
 
def run
now = Time.now.to_f
sleep now.ceil - now + 0.001
 
every(1) { notify 'time_change', Time.now.inspect }
info 'Registered to send time_change'
 
every(10) { notify 'keepalive', Time.now.to_i }
info 'Registered to send keepalive'
end
end
 
class TimeClient
include PGNotifications
include Celluloid::Logger
 
KEEPALIVE_LIMIT = 30 # keep it short
 
def initialize(socket)
info "Streaming time changes to client"
@socket = socket
@limit = Time.now.to_i + KEEPALIVE_LIMIT
 
async.start_listening
async.listen('time_change', :notify_time_change)
async.listen('keepalive', :report_keepalive)
end
 
def notify_time_change(channel, new_time)
@socket << new_time
rescue Reel::SocketError
info "Time client disconnected"
unlisten(channel)
terminate
end
 
def report_keepalive(channel, timestamp)
info "Current keepalive timestamp: #{timestamp}"
# unlisten(channel) if timestamp.to_i > @limit
end
end
 
class WebServer < Reel::Server::HTTP
include Celluloid::Logger
 
def initialize(host = "127.0.0.1", port = 1234)
info "Time server example starting on #{host}:#{port}"
super(host, port, &method(:on_connection))
end
 
def on_connection(connection)
connection.each_request do |request|
if request.websocket?
info "Received a WebSocket connection"
handle_websocket request.websocket
else
handle_request request
end
end
end
 
def handle_request(request)
if request.url == "/"
return render_index(request)
end
 
info "404 Not Found: #{request.path}"
request.respond :not_found, "Not found"
end
 
def handle_websocket(socket)
if socket.url == "/timeinfo"
TimeClient.new(socket)
else
info "Received invalid WebSocket request for: #{socket.url}"
socket.close
end
end
 
def render_index(request)
info "200 OK: /"
request.respond :ok, <<-HTML
<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<title>Reel WebSockets time server example</title>
<style>
body {
font-family: "HelveticaNeue-Light", "Helvetica Neue Light", "Helvetica Neue", Helvetica, Arial, "Lucida Grande", sans-serif;
font-weight: 300;
text-align: center;
}
#content {
width: 800px;
margin: 0 auto;
background: #EEEEEE;
padding: 1em;
}
</style>
</head>
<script>
var SocketKlass = "MozWebSocket" in window ? MozWebSocket : WebSocket;
var ws = new SocketKlass('ws://' + window.location.host + '/timeinfo');
ws.onmessage = function(msg){
document.getElementById('current-time').innerHTML = msg.data;
}
</script>
<body>
<div id="content">
<h1>Time Server Example</h1>
<div>The time is now: <span id="current-time">...</span></div>
</div>
</body>
</html>
HTML
end
end
 
TimeServer.supervise_as :time_server
WebServer.supervise_as :reel
 
sleep
Owner
tpitale commented

Works now with the latest reel at this time (0.5.0).

Seems to only be limited by the available number of postgresql connections.

Owner
tpitale commented

Revision 14eeea263bb0320c2cedaee36ec6d792e9b1e492 now supports multiple listens per actor.

I've recreated the wait_for_notify from the pg gem and enhanced it to work with Celluloid::IO.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Something went wrong with that request. Please try again.