RFC: Synchronous updates by default

src/connection.jl

@@ -4,77 +4,10 @@ export AbstractConnection
 
 abstract type AbstractConnection end
 
-"""
-The maximum number of messages to allow into the outbox.
-"""
-const DEFAULT_OUTBOX_LIMIT = 32
-
-"""
-    ConnectionPool([outbox[, connections]])
-
-Manages the distribution of messages from the `outbox` channel to a set of
-connections. The ConnectionPool asynchronously takes messages from its outbox
-and sends each message to every connection in the pool. Any closed connections
-are automatically removed from the pool.
-"""
-struct ConnectionPool
-    outbox::Channel
-    connections::Set{AbstractConnection}
-    condition::Condition
-end
-
-function ConnectionPool(
-        outbox::Channel = Channel{Any}(DEFAULT_OUTBOX_LIMIT),
-        connections=Set{AbstractConnection}(),
-)
-    pool = ConnectionPool(
-        outbox,
-        connections,
-        Condition(),
-    )
-
-    # Catch errors here, otherwise they are lost to the void.
-    @async try
-        process_messages(pool)
-    catch exc
-        @error(
-            "An error ocurred in the while processing messages from a "
-                * "ConnectionPool.",
-            exception=exc,
-        )
-    end
-
-    return pool
+function sendall(pool::Set, msg)
+    process_messages(pool, msg)
 end
 
-function addconnection!(pool::ConnectionPool, conn::AbstractConnection)
-    push!(pool.connections, conn)
-    notify(pool)
-end
-
-function Sockets.send(pool::ConnectionPool, msg)
-    if length(pool.outbox.data) >= pool.outbox.sz_max
-        # TODO: https://github.com/JuliaGizmos/WebIO.jl/issues/343
-        return
-    end
-    put!(pool.outbox, msg)
-end
-
-"""
-    ensure_connection(pool)
-
-Ensure that the pool has at least one connection, potentially blocking the
-current task until that is the case. Also processes incoming connections.
-"""
-function ensure_connection(pool::ConnectionPool)
-    if isempty(pool.connections)
-        wait(pool.condition)
-    end
-end
-
-Base.wait(pool::ConnectionPool) = ensure_connection(pool)
-Base.notify(pool::ConnectionPool) = notify(pool.condition)
-
 """
     process_messages(pool)
 
@@ -83,16 +16,14 @@ frontends.
 
 This function should be run as a task (it will block forever otherwise).
 """
-function process_messages(pool::ConnectionPool)
-    ensure_connection(pool)
-    while true
-        msg = take!(pool.outbox)
+function process_messages(pool::Set, msg)
+    if !isempty(pool)
         @sync begin
             # This may result in sending to no connections, but we're okay with
             # that because we'll just get the next value of the observable
             # (messages are fire and forget - WebIO makes no guarantees that
             # messages are ever actually delivered).
-            for connection in pool.connections
+            for connection in pool
                 @async send_message(pool, connection, msg)
             end
         end
@@ -106,7 +37,7 @@ Send a message to an individual connection within a pool, handling errors and
 deleting the connection from the pool if necessary.
 """
 function send_message(
-        pool::ConnectionPool,
+        pool::Set,
         connection::AbstractConnection,
         msg,
 )::Nothing
@@ -115,7 +46,7 @@ function send_message(
             send(connection, msg)
         else
             @info "Connection is not open." connection
-            delete!(pool.connections, connection)
+            delete!(pool, connection)
         end
     catch ex
         @error(

src/messaging.jl

@@ -92,7 +92,7 @@ function dispatch_command(conn::AbstractConnection, data)
         if cmd == "_setup_scope"
             @warn("Client used deprecated command: _setup_scope.", maxlog=1)
         end
-        addconnection!(scope.pool, conn)
+        push!(scope.pool, conn)
     elseif cmd == "update_observable"
         if !haskey(data, "name")
             @error "update_observable message missing \"name\" key."

src/scope.jl

@@ -11,8 +11,7 @@ export Scope,
        onimport,
        ondependencies,
        adddeps!,
-       import!,
-       addconnection!
+       import!
 
 import Compat.Sockets: send
 import Observables: Observable, AbstractObservable, listeners
@@ -34,7 +33,7 @@ mutable struct Scope
     # where each JS-string is a function that is invoked when the observable
     # changes.
     jshandlers::Any
-    pool::ConnectionPool
+    pool::Set{AbstractConnection}
 
     mount_callbacks::Vector{JSString}
 
@@ -45,7 +44,7 @@ mutable struct Scope
             systemjs_options::Any,
             imports::Vector{Asset},
             jshandlers::Any,
-            pool::ConnectionPool,
+            pool,
             mount_callbacks::Vector{JSString}
         )
         scope = new(
@@ -119,7 +118,6 @@ myscope = Scope(
 """
 function Scope(;
         dom = dom"span"(),
-        outbox::Union{Channel, Nothing} = nothing,
         observs::Dict = ObsDict(),
         private_obs::Set{String} = Set{String}(),
         systemjs_options = nothing,
@@ -137,7 +135,7 @@ function Scope(;
         )
     end
     imports = Asset[Asset(i) for i in imports]
-    pool = outbox !== nothing ? ConnectionPool(outbox) : ConnectionPool()
+    pool = Set{AbstractConnection}()
     return Scope(
         dom, observs, private_obs, systemjs_options,
         imports, jshandlers, pool, mount_callbacks
@@ -153,7 +151,6 @@ function Scope(id; kwargs...)
 end
 
 (w::Scope)(arg) = (w.dom = arg; w)
-Base.wait(scope::Scope) = ensure_connection(scope.pool)
 
 function Observables.on(f, w::Scope, key)
     key = string(key)
@@ -277,7 +274,7 @@ function send_command(scope::Scope, command, data::Pair...)
         "scope" => scopeid(scope),
         data...
     )
-    send(scope.pool, message)
+    sendall(scope.pool, message)
     nothing
 end
 
@@ -369,7 +366,7 @@ function ensure_sync(ctx, key)
     ob = ctx.observs[key][1]
     # have at most one synchronizing handler per observable
     if !any(x->isa(x, SyncCallback) && x.ctx==ctx, listeners(ob))
-        f = SyncCallback(ctx, (msg) -> send_update_observable(ctx, key, msg))
+        f = (msg) -> send_update_observable(ctx, key, msg)
         on(SyncCallback(ctx, f), ob)
     end
 end