whimsical-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ru...@apache.org
Subject [whimsy] branch master updated: board/agenda: s/EventSource/WebSocket/
Date Wed, 12 Oct 2016 14:34:06 GMT
This is an automated email from the ASF dual-hosted git repository.

rubys pushed a commit to branch master
in repository https://git-dual.apache.org/repos/asf/whimsy.git

The following commit(s) were added to refs/heads/master by this push:
       new  117473a   board/agenda: s/EventSource/WebSocket/
117473a is described below

commit 117473a0f00b6f3ef60f573c381e7d2c20ace353
Author: Sam Ruby <rubys@intertwingly.net>
AuthorDate: Wed Oct 12 10:33:24 2016 -0400

    board/agenda: s/EventSource/WebSocket/
---
 www/board/agenda/daemon/session.rb                |  23 ++--
 www/board/agenda/daemon/wss.rb                    |  49 +++++---
 www/board/agenda/main.rb                          |   4 +-
 www/board/agenda/models/events.rb                 | 124 --------------------
 www/board/agenda/models/ipc.rb                    | 134 +++++-----------------
 www/board/agenda/public/assets/eventsource.min.js |   6 -
 www/board/agenda/routes.rb                        |  24 ----
 www/board/agenda/views/models/events.js.rb        |  53 +++++----
 www/board/agenda/views/sw.js.rb                   |  29 +----
 9 files changed, 113 insertions(+), 333 deletions(-)

diff --git a/www/board/agenda/daemon/session.rb b/www/board/agenda/daemon/session.rb
index 29d70b0..7e99dfd 100644
--- a/www/board/agenda/daemon/session.rb
+++ b/www/board/agenda/daemon/session.rb
@@ -3,6 +3,8 @@ require 'thread'
 require 'securerandom'
 require 'concurrent'
 
+require 'whimsy/asf/config'
+
 #
 # Low-tech, file based session manager.  Each session is stored as a separate
 # file on disk, and expires after two days.  Each request for a new session
@@ -18,6 +20,7 @@ require 'concurrent'
 #
 
 class Session
+  AGENDA_WORK = ASF::Config.get(:agenda_work).untaint || '/srv/agenda'
   WORKDIR = File.expand_path('sessions', AGENDA_WORK)
   DAY = 24*60*60 # seconds
 
@@ -70,16 +73,19 @@ class Session
         secret = File.basename(file)
         session = @@sessions[secret]
 
-        File.delete file if session and session[:mtime] < Time.now - 2 * DAY
 
         if File.exist? file
-          # update class variables if the file changed
-          mtime = File.mtime(file)
-          next if session and session[:mtime] == mtime
+          if File.mtime(file) < Time.now - 2 * DAY
+            File.delete file 
+          else
+            # update class variables if the file changed
+            mtime = File.mtime(file)
+            next if session and session[:mtime] == mtime
 
-          session = {id: File.read(file), secret: secret, mtime: mtime}
-          @@sessions[secret] == session
-          @@users[session[:id]] << session
+            session = {id: File.read(file), secret: secret, mtime: mtime}
+            @@sessions[secret] = session
+            @@users[session[:id]] << session
+          end
         else
           # remove session if the file no longer exists
           @@users[session[:id]].delete(session) if session
@@ -91,4 +97,7 @@ class Session
 
   # ensure the working directory exists
   FileUtils.mkdir_p WORKDIR
+
+  # load initial data from disk
+  self.load
 end
diff --git a/www/board/agenda/daemon/wss.rb b/www/board/agenda/daemon/wss.rb
index ba3c205..2878736 100755
--- a/www/board/agenda/daemon/wss.rb
+++ b/www/board/agenda/daemon/wss.rb
@@ -7,7 +7,10 @@ require 'optparse'
 require 'yaml'
 require 'rbconfig'
 
-clients = []
+require_relative './session'
+
+users = {}
+sessions = {}
 
 ########################################################################
 #                         Parse argument list                          #
@@ -85,7 +88,7 @@ listener.start
 ########################################################################
 
 at_exit do
-  clients.each do |client|
+  sessions.keys.each do |client|
     client.close
   end
 end
@@ -106,13 +109,9 @@ end
 
 EM.run do
   WebSocket::EventMachine::Server.start(server_options) do |ws|
-    ws.onopen do |handshake|
-      ws.comm_inactivity_timeout = options.timeout
-      clients << ws
-    end
-
     ws.onclose do 
-     clients.delete ws
+      id = sessions.delete(ws)
+      users[id].delete ws if id
     end
 
     ws.onmessage do |msg|
@@ -120,13 +119,33 @@ EM.run do
       headers = msg.slice!(/\A(\w+:\s*.*\r?\n)\s*(\n|\Z)/).to_s
       headers = YAML.safe_load(headers) || {} rescue {}
 
-      # echo message to all of the clients
-      clients.each do |client|
-        EM.defer(
-          ->() {client.send msg},
-          ->(response) {},
-          ->(error) {client.close rescue nil}
-        )
+      if headers['session']
+        STDERR.puts headers.inspect
+        session = Session[headers['session']]
+        STDERR.puts session.inspect
+        if session
+          users[session[:id]] = ws
+          sessions[ws] = session[:id]
+        end
+      end
+
+      # forward message
+      unless msg.empty?
+        if headers['private']
+          # send only to a specific user
+          clients = users[headers['private']] || []
+        else
+          # send to all users
+          clients = sessions.keys
+        end
+
+        clients.each do |client|
+          EM.defer(
+            ->() {client.send msg},
+            ->(response) {},
+            ->(error) {client.close rescue nil}
+          )
+        end
       end
     end
   end
diff --git a/www/board/agenda/main.rb b/www/board/agenda/main.rb
index 2cf62c3..018a914 100755
--- a/www/board/agenda/main.rb
+++ b/www/board/agenda/main.rb
@@ -8,7 +8,6 @@ require 'whimsy/asf/agenda'
 
 require 'wunderbar/sinatra'
 require 'wunderbar/react'
-require 'wunderbar/eventsource'
 require 'wunderbar/bootstrap/theme'
 require 'ruby2js/filter/functions'
 require 'ruby2js/filter/require'
@@ -36,13 +35,14 @@ FileUtils.mkdir_p AGENDA_WORK if not Dir.exist? AGENDA_WORK
 require_relative './routes'
 require_relative './models/ipc'
 require_relative './models/pending'
-require_relative './models/events'
 require_relative './models/agenda'
 require_relative './models/minutes'
 require_relative './models/comments'
 require_relative './helpers/string'
 require_relative './daemon/session'
 
+require 'websocket-client-simple'
+
 # if AGENDA_WORK doesn't exist yet, make it
 if not Dir.exist? AGENDA_WORK
   require 'fileutils'
diff --git a/www/board/agenda/models/events.rb b/www/board/agenda/models/events.rb
deleted file mode 100644
index 412fd29..0000000
--- a/www/board/agenda/models/events.rb
+++ /dev/null
@@ -1,124 +0,0 @@
-#
-# A centralized subscription service for server side events
-#  * Sends a heartbeat every 25 seconds
-#  * Closes all sockets when restart is detected
-#
-
-class EventService
-  attr_accessor :user
-  attr_accessor :token
-  attr_accessor :queue
-
-  @@subscriptions = {}
-  @@restart = false
-  @@next_token = 1
-  @@cache = Hash.new(mtime: 0)
-
-  # key/value store (for agenda purposes)
-  def self.[](file)
-    @@cache[file]
-  end
-
-  def self.[]=(file, data)
-    @@cache[file] = data
-  end
-
-  # create a new subscription
-  def self.subscribe(user)
-    self.hook_restart
-    present = EventService.present
-    subscriber = EventService.new(user)
-    @@subscriptions[subscriber.token] = subscriber
-    if not present.include? user
-      EventService.post type: :arrive, user: user, 
-        present: EventService.present, timestamp: Time.now.to_f*1000
-    end
-    subscriber.token
-  end
-
-  # post an event to all subscribers
-  def self.post(event)
-    return unless event
-
-    @@subscriptions.each do |token, subscriber|
-      if
-        not Hash === event or not event[:private] or # broadcast
-        event[:private] == subscriber.user           # narrowcast
-      then
-        subscriber.queue << event
-      end
-    end
-    event
-  end
-
-  # list of users present
-  def self.present
-    @@subscriptions.map {|token, subscriber| subscriber.user}.uniq.sort
-  end
-
-  # capture user information associated with this queue
-  def initialize(user)
-    @user = user
-    @token = @@next_token
-    @queue = Queue.new
-    @@next_token += 1
-    super()
-  end
-
-  def self.pop(token)
-    subscription = @@subscriptions[token]
-    subscription.queue.pop if subscription
-  end
-
-  # remove a subscription
-  def self.unsubscribe(token)
-    event = @@subscriptions.delete token
-    present = EventService.present
-    if event and not present.include? event.user
-      EventService.post type: :depart, user: event.user, present: present,
-        timestamp: Time.now.to_f*1000
-    end
-  end
-
-  # When restart signal is detected, close all open connections
-  def self.hook_restart
-    # puma uses SIGUSR2
-    restart_usr2 ||= trap 'SIGUSR2' do
-      restart_usr2.call if Proc === restart_usr2
-      begin
-        EventService.post(:exit)
-      rescue ThreadError
-        # some versions of Ruby don't allow queue operations in traps
-        @restart = true
-      end
-    end
-
-    # thin uses SIGHUP
-    restart_hup ||= trap 'SIGHUP' do
-      restart_hup.call if Proc === restart_hup
-      begin
-        EventService.post(:exit)
-      rescue ThreadError
-        # some versions of Ruby don't allow queue operations in traps
-        @restart = true
-      end
-    end
-  end
-
-  # As some TCP/IP implementations will close idle sockets after as little
-  # as 30 seconds, sent out a heartbeat every 25 seconds.  Due to limitations
-  # of some versions of Ruby (2.0, 2.1), this is lowered to every 5 seconds
-  # in development mode to allow for quicker restarting after a trap/signal.
-  Thread.new do
-    loop do
-      sleep(ENV['RACK_ENV'] == 'development' ? 5 : 25)
-
-      if @restart
-        EventService.post(:exit)
-        @restart = false
-      else
-        EventService.post(:heartbeat)
-      end
-    end
-  end
-end
diff --git a/www/board/agenda/models/ipc.rb b/www/board/agenda/models/ipc.rb
index ecc6f62..a87da8c 100644
--- a/www/board/agenda/models/ipc.rb
+++ b/www/board/agenda/models/ipc.rb
@@ -1,117 +1,37 @@
-#
-# IPC server based on Ruby's DRB.  Manages publish and subscribe queues as
-# well as distributed hash.
-#
-# Key principles:
-#   * Reserves a 9146 as a socket
-#
-#   * All IPC server operations are provided by a single class, and all server
-#     operations are class methods that return primitive (round-trippable to
-#     JSON) objects.  This provides the IPC_Server logic the ability to
-#     consistently capture and recover from DRBConnErrors.
-#
-#   * No I/O or computational intensive processing is provided by the IPC
-#     server.  
-#
+module IPC
 
-require 'drb'
-require 'thread'
-
-class IPC_Server
-  SOCKET = 'druby://:9146'
-
-  attr_accessor :object
-
-  def initialize(object)
-    IPC_Server.start_server
-    @object = object
-  end
-
-  def method_missing(method, *args, &block)
-    loop do
-      begin
-        return @object.send method, *args, &block
-      rescue DRb::DRbConnError => e
-        IPC_Server.start_server
-        sleep 0.1
-      end
-    end
-  end
-
-  def self.start_server
-    pid = fork do
-      # run server code in a separate process
-      ruby = File.join(
-        RbConfig::CONFIG["bindir"],
-        RbConfig::CONFIG["ruby_install_name"] + RbConfig::CONFIG["EXEEXT"]
-      )
-
-      exec(ruby, __FILE__.dup.untaint, '--server-only')
-    end
-
-    Process.detach pid
-
-    at_exit {Process.kill 'INT', pid rescue nil}
-  end
-
-  ENV['RACK_ENV'] ||= 'development'
-end
-
-# launch server or client
-if ENV['RACK_ENV'] == 'test'
-
-  require_relative 'events'
-  IPC = EventService
-
-elsif ARGV[0] == '--server-only'
-
-  if  __FILE__ == $0
-    Signal.trap('INT') {sleep 1; exit}
-    Signal.trap('TERM') {exit}
-    Signal.trap('USR2') {exit}
-
-    require_relative 'events'
-
-    begin
-      DRb.start_service(IPC_Server::SOCKET, EventService)
-      DRb.thread.join
-    rescue Errno::EADDRINUSE => e
-      exit
-    end
+  if Dir.exist? '/etc/letsencrypt'
+    @@url ="wss://127.0.0.1:34234"
+  else
+    @@url ="ws://127.0.0.1:34234"
   end
 
-else
+  def self.post(data)
+    thread = Thread.new do
+      # post to web socket server
+      ws = WebSocket::Client::Simple.connect @@url
 
-  # IPC client
-  IPC = IPC_Server.new(DRbObject.new(nil, IPC_Server::SOCKET))
-
-end
-
-# For demonstration / debugging purposes
-if __FILE__ == $0
-  require 'etc'
-  user = ARGV.pop || Etc.getlogin
-
-  queue = IPC.subscribe(user)
-
-  at_exit do
-    IPC.unsubscribe(queue)
-  end
-
-  Thread.new do
-    loop do
-      event = IPC.pop(queue)
-      if event
-        puts '>> ' + event.inspect
-      else
-        queue = IPC.subscribe(user)
+      begin
+        done = false
+        ws.on :open do
+          if data[:private]
+            headers = "session: #{data[:private]}\n\n"
+          else
+            headers = ''
+          end
+
+          ws.send headers + JSON.dump(data)
+          done = true
+        end
+
+        sleep 0.1 until done
+      ensure
+        ws.close
       end
     end
   end
 
-  loop do
-    data = gets.strip
-    exit if data.empty?
-    IPC.post data
+  def self.present
+    [] # TBD
   end
 end
diff --git a/www/board/agenda/public/assets/eventsource.min.js b/www/board/agenda/public/assets/eventsource.min.js
deleted file mode 100644
index 5cacdc6..0000000
--- a/www/board/agenda/public/assets/eventsource.min.js
+++ /dev/null
@@ -1,6 +0,0 @@
-/** @license
- * eventsource.js
- * Available under MIT License (MIT)
- * https://github.com/Yaffle/EventSource/
- */
-!function(a){"use strict";function b(){this.data={}}function c(){this.listeners=new b}function
d(a){setTimeout(function(){throw a},0)}function e(a){this.type=a,this.target=void 0}function
f(a,b){e.call(this,a),this.data=b.data,this.lastEventId=b.lastEventId}function x(a,b){var
c=a;return c!==c&&(c=b),v>c?v:c>w?w:c}function y(a,b,c){try{"function"==typeof
b&&b.call(a,c)}catch(e){d(e)}}function z(b,d){function P(){H=o,void 0!==D&&(D.abort(),D=void
0),0!==E&&(clearTimeout(E),E=0),0!==F&&(cl [...]
diff --git a/www/board/agenda/routes.rb b/www/board/agenda/routes.rb
index 5ab1e82..e77d738 100755
--- a/www/board/agenda/routes.rb
+++ b/www/board/agenda/routes.rb
@@ -234,30 +234,6 @@ get '/json/historical-comments' do
   _json HistoricalComments.comments
 end
 
-# event stream for server sent events (a.k.a EventSource)
-get '/events', provides: 'text/event-stream' do
-  stream :keep_open do |out|
-    subscription = IPC.subscribe(env.user)
-    out.callback {IPC.unsubscribe(subscription)}
-
-    loop do
-      event = IPC.pop(subscription)
-      if Hash === event or Array === event
-        out << "data: #{JSON.dump(event)}\n\n"
-      elsif event == :heartbeat
-        out << ":\n"
-      elsif event == :exit
-        out.close
-        break
-      elsif event == nil
-        subscription = IPC.subscribe(env.user)
-      else
-        out << "data: #{event.inspect}\n\n"
-      end
-    end
-  end
-end
-
 # draft minutes
 get '/text/draft/:file' do |file|
   agenda = "board_agenda_#{file.gsub('-','_')}.txt".untaint
diff --git a/www/board/agenda/views/models/events.js.rb b/www/board/agenda/views/models/events.js.rb
index 351b8ae..cbad860 100644
--- a/www/board/agenda/views/models/events.js.rb
+++ b/www/board/agenda/views/models/events.js.rb
@@ -1,7 +1,7 @@
 #
-# Motivation: browsers limit the number of open HTTP connections to any
-# one host to somewhere between 4-10.  "Long polling" keeps an HTTP
-# connection open making it impractical to have one EventSource per tab.
+# Motivation: browsers limit the number of open web socket connections to any
+# one host to somewhere between 6 and 250, making it impractical to have one
+# Web Socket per tab.
 #
 # The solution below uses localStorage to communicate between tabs, with
 # the majority of logic involved with the "election" of a master.  This
@@ -9,18 +9,10 @@
 #
 # Alternatives include: 
 #
-# * Replacing Server Side Events with Web Sockets which will require more open
-#   connections (and therefore server load) as well as require special proxy
-#   configuration (mod_proxy_wstunnel) and server coding (faye-websocket or
-#   equivalent).
-#
 # * Replacing localStorage with Service Workers.  This would be much cleaner,
 #   unfortunately Service Workers aren't widely deployed yet.  Sadly, the
 #   state isn't much better for Shared Web Workers.
 #
-# Notable downside to Server Side Events is lack of native support by IE.
-# This is readily addressed by a polyfill.
-#
 ###
 #
 # Class variables:
@@ -33,6 +25,7 @@
 
 class Events
   @@subscriptions = {}
+  @@socket = nil
 
   def self.subscribe event, &block
     @@subscriptions[event] ||= []
@@ -109,23 +102,43 @@ class Events
 
   # master logic
   def self.master()
-    events = EventSource.new('../events')
+    self.connectToServer()
 
-    # dispatch events received to all windows
-    events.addEventListener :message do |event|
-      localStorage.setItem("#{@@prefix}-event", event.data)
-      self.dispatch event.data
-    end
-
-    # proof of life
+    # proof of life; maintain connection to the server
     setTimeout 25_000 do
       localStorage.setItem("#{@@prefix}-timestamp", Date.new().getTime())
+      self.connectToServer()
     end
 
     # close connection on exit
     window.addEventListener :unload do |event|
-      events.close()
+      @@socket.close() if @@socket
+    end
+  end
+
+  # establish a connection to the server
+  def self.connectToServer()
+    return if @@socket
+
+    socket_url = window.location.protocol.sub('http', 'ws') + "//" + 
+        window.location.hostname + ':34234/'
+
+    @@socket = WebSocket.new(socket_url)
+
+    def @@socket.onopen(event)
+      @@socket.send "session: #{Server.session}\n\n"
+    end
+
+    def @@socket.onmessage(event)
+      localStorage.setItem("#{@@prefix}-event", event.data)
+      self.dispatch event.data
+    end
+
+    def @@socket.onclose(event)
+      @@socket = nil
     end
+  rescue => e
+    console.log e
   end
 
   # dispatch logic (common to all tabs)
diff --git a/www/board/agenda/views/sw.js.rb b/www/board/agenda/views/sw.js.rb
index d8876c7..7100d65 100644
--- a/www/board/agenda/views/sw.js.rb
+++ b/www/board/agenda/views/sw.js.rb
@@ -1,38 +1,11 @@
 #
 # A very simple service worker
 #
-#   1) Create an event source and pass all events received to all clients
-#
-#   2) Return back cached bootstrap page instead of fetching agenda pages
+#   *) Return back cached bootstrap page instead of fetching agenda pages
 #      from the network.  Bootstrap will construct page from cached
 #      agenda.json, as well as update the cache.
 # 
 
-events = nil
-
-self.addEventListener :activate do |event|
-  # close any pre-existing event socket
-  if events
-    begin
-      events.close()
-    rescue => e
-      events = nil
-    end
-  end
- 
-  # create a new event source
-  events = EventSource.new('events')
- 
-  # dispatch any events received to clients
-  events.addEventListener :message do |event|
-    clients.matchAll().then do |list|
-      list.each do |client|
-        client.postMessage(event.data)
-      end
-    end
-  end
-end
-
 self.addEventListener :fetch do |event|
   scope = self.registration.scope
   url = event.request.url

-- 
To stop receiving notification emails like this one, please contact
['"commits@whimsical.apache.org" <commits@whimsical.apache.org>'].

Mime
View raw message