Allow "retry: false" to fail and pass exceptions to caller

1 files changed, 143 insertions, 152 deletions

diff --git a/src/obswebsocket.cr b/src/obswebsocket.cr
index 6f623a0..0daea20 100644
--- a/src/obswebsocket.cr
+++ b/src/obswebsocket.cr
@@ -29,9 +29,6 @@ module OBS
     return( request.to_s )
   end
 
-  class WriterUnavailableException < Exception
-  end
-
   class PasswordMissingException < Exception
   end
 
@@ -42,28 +39,34 @@ module OBS
   end
 
   class WebSocket
-    @@writer : Fiber | Nil
-    @@reqchan = Channel( Tuple( ( Channel( JSON::Any ) | Nil ), String ) ).new
-    @scenecollection = OBS::SceneCollection.new( @@reqchan, @@inputs )
-    @@inputs = OBS::Inputs.new( @@reqchan )
-    @@scenes = OBS::SceneList.new( @@reqchan, @@inputs )
-    @sources = OBS::Sources.new( @@reqchan, @@inputs, @@scenes )
-    @outputs = OBS::Outputs.new( @@reqchan )
-    @video = OBS::VideoSettings.new( @@reqchan )
-    @stats = OBS::Stats.new( @@reqchan )
+    # A bunch of these get initialized in OBS::WebSocket.initialize,
+    # but we still want a class-wide typedef here to avoid managing Nils
+    @@inputs         : OBS::Inputs          = OBS::Inputs.allocate
+    @@scenes         : OBS::SceneList       = OBS::SceneList.allocate
+    @sources         : OBS::Sources         = OBS::Sources.allocate
+    @scenecollection : OBS::SceneCollection = OBS::SceneCollection.allocate
+    @outputs         : OBS::Outputs         = OBS::Outputs.allocate
+    @video           : OBS::VideoSettings   = OBS::VideoSettings.allocate
+    @stats           : OBS::Stats           = OBS::Stats.allocate
+    @@eventsubs = Array( Channel(JSON::Any) ).new
+    @@openrequests = Hash( String, Channel( JSON::Any ) ).new
     @negotiated = false
+    @@negotiationdelayqueue = 0
+    @@negotiationdelay = Channel( Exception | Nil ).new
     @connecterror = 0
     @shutdown = false
     @closed = true
     def closed?
       return @closed
     end
+    def eventsub_add( channel : Channel( JSON::Any ) )
+      # FIXME: this currently subscribes to everything
+      # maybe make it more specific at some point
+      @@eventsubs.push( channel )
+    end
     def negotiated?
       return @negotiated
     end
-    def reqchan
-      return @@reqchan
-    end
     def scenecollection
       return @scenecollection
     end
@@ -92,18 +95,32 @@ module OBS
     end
     # request channel, response channel
     def send( json : String )
-      if ( @@writer && ! @@writer.not_nil!.dead? )
-        @@reqchan.send( { nil, json } )
-      else
-        raise WriterUnavailableException.new( "Writer fiber is not available." )
+      unless @negotiated
+        STDERR.puts( "DELAYING" )
+        @@negotiationdelayqueue += 1
+        negotiationresult = @@negotiationdelay.receive
+        if negotiationresult.is_a?( Exception )
+          raise negotiationresult
+        end
+        @negotiated || return false # not sure why this condition would occur without an exception
       end
+      STDERR.puts( "SENT: #{JSON.parse(json).pretty_inspect}" )
+      @@obs_pubsub && @@obs_pubsub.not_nil!.send( json )
     end
     def send( reschan : ( Channel( JSON::Any ) | Nil ), json : String )
-      if @@writer && ! @@writer.not_nil!.dead?
-        @@reqchan.send( { reschan, json } )
-      else
-        raise WriterUnavailableException.new( "Writer fiber is not available." )
+      unless @negotiated
+        STDERR.puts( "DELAYING" )
+        @@negotiationdelayqueue += 1
+        negotiationresult = @@negotiationdelay.receive
+        if negotiationresult.is_a?( Exception )
+          raise negotiationresult
+        end
+        @negotiated || return false # not sure why this condition would occur without an exception
       end
+      STDERR.puts( "SENT: #{JSON.parse(json).pretty_inspect}" )
+      @@obs_pubsub && @@obs_pubsub.not_nil!.send( json )
+      @@openrequests[ JSON.parse( json )["d"]["requestId"].as_s ] = reschan
+      # maybe process reschan.receive here?
     end
     def request( type : String, id : String, data : ( String | Nil | JSON::Any ) = nil )
       request = JSON.build do |json|
@@ -120,7 +137,7 @@ module OBS
           end
         end
       end
-      self.reqchan.send( { nil, request.to_s } )
+      self.send( request.to_s )
     end
     def request( reschan : ( Channel( JSON::Any ) | Nil ), type : String, id : String, data : ( String | Nil | JSON::Any ) = nil )
       request = JSON.build do |json|
@@ -137,65 +154,33 @@ module OBS
           end
         end
       end
-      self.reqchan.send( { reschan, request.to_s } )
+      self.send( reschan, request.to_s )
     end
     def initialize( uri : String, password : String | Nil = nil, retry : Bool = true )
+
+      @scenecollection.initialize( self )
+      @@inputs.initialize( self )
+      @@scenes.initialize( self, @@inputs )
+      @sources.initialize( self, @@inputs, @@scenes )
+      @outputs.initialize( self )
+      @video.initialize( self )
+      @stats.initialize( self )
+
       spawn do
-        eventsubs = Array( Channel(JSON::Any) ).new # initialize this here to it survives across connection retries
         loop do
           if @shutdown
             break
           end
-          @obs_pubsub = HTTP::WebSocket.new( URI.parse( uri ), HTTP::Headers{"Cookie" => "SESSIONID=1235", "Sec-WebSocket-Protocol" => "OBS.json"} )
-          @obs_pubsub.on_close do | message | # for some reason HTTP::WebSocket.closed? SIGSEGVs on a dangling pointer on the 1.6 runtime, so we hack around it with our own state variable for now.
+          # FFS why is HTTP::WebSocket.initialize protected?
+          # Now we have to .not_nil! this shit everywhere
+          # @@obs_pubsub.initialize( HTTP::WebSocket::Protocol.new( URI.parse( uri ), HTTP::Headers{"Cookie" => "SESSIONID=1235", "Sec-WebSocket-Protocol" => "OBS.json"}  ) )
+          @@obs_pubsub = HTTP::WebSocket.new( URI.parse( uri ), HTTP::Headers{"Cookie" => "SESSIONID=1235", "Sec-WebSocket-Protocol" => "OBS.json"}  )
+          @@obs_pubsub.not_nil!.on_close do | message | # for some reason HTTP::WebSocket.closed? SIGSEGVs on a dangling pointer on the 1.6 runtime, so we hack around it with our own state variable for now.
             @closed = true
           end
           @closed = false
-          openrequests = Hash( String, Channel( JSON::Any ) ).new
           #metachan = Channel( Tuple( String, Channel( JSON::Any ) ) ).new
-          @@writer = spawn do
-            queue = Array( Tuple( Channel( JSON::Any ) | Nil, String ) ).new
-            while msgtuple = @@reqchan.receive
-              if msgtuple[1] == "clear queue"
-                queue.each do | msg |
-                  reschan = msg[0]
-                  STDERR.print( "SENT: " )
-                  json = JSON.parse(msg[1])
-                  STDERR.print( json.pretty_inspect )
-                  if reschan
-                    openrequests[ json["d"]["requestId"].as_s ] = reschan
-                  end
-                  @obs_pubsub.send( msg[1] )
-                end
-              elsif msgtuple[1] == "break"
-                break
-              elsif msgtuple[1] == "subscribe events"
-                reschan = msgtuple[0]
-                if reschan
-                  eventsubs.push(reschan)
-                end
-              elsif @negotiated
-                reschan = msgtuple[0]
-                STDERR.print( "SENT: " )
-                json = JSON.parse( msgtuple[1] )
-                STDERR.print( json.pretty_inspect )
-                if reschan
-                  openrequests[ json["d"]["requestId"].as_s ] = reschan
-                end
-                @obs_pubsub.send( msgtuple[1] )
-              elsif JSON.parse( msgtuple[1] )["op"].as_i64 == 1
-                STDERR.print( "SENT: " )
-                json = JSON.parse( msgtuple[1] )
-                STDERR.print( json.pretty_inspect )
-                @obs_pubsub.send( msgtuple[1] )
-              else
-                STDERR.print( "QUEUED: " )
-                STDERR.print( msgtuple[1].pretty_inspect )
-                queue.push( { msgtuple[0], msgtuple[1] } )
-              end
-            end
-          end
-          @obs_pubsub.on_message do | message |
+          @@obs_pubsub.not_nil!.on_message do | message |
             json = JSON.parse( message )
             STDERR.print( "RECEIVED: ")
             STDERR.print( json.to_pretty_json )
@@ -227,14 +212,17 @@ module OBS
                 end
               end
 
-              self.send( hello.to_s )
+              @@obs_pubsub.not_nil!.send( hello.to_s )
             when 2 # identified
-                @negotiated = true
-                @connecterror = 0
-                self.reqchan.send( { nil, "clear queue" } )
+              @negotiated = true
+              @@negotiationdelayqueue.times do
+                @@negotiationdelay.send( nil )
+                @@negotiationdelayqueue -= 1
+              end
+              @connecterror = 0
             when 5 # event
               # FIXME: These should be switched over to Enum flags
-              eventsubs.each do | eventchan |
+              @@eventsubs.each do | eventchan |
                 eventchan.send( json["d"] )
               end
               # A Fiber.yield occurs immediately after this to make sure "json" doesn't get overwritten before we can use it.
@@ -248,11 +236,9 @@ module OBS
                   edata = d["eventData"]
                   @@scenes.programcache( edata["sceneName"].as_s )
                 when "CurrentSceneCollectionChanging"
-                  # FIXME: switch to queue in reqchan
                   @scenecollection.deletecache()
                   @@scenes.deletecache()
                 when "CurrentSceneCollectionChanged"
-                  # FIXME: empty queue in reqchan
                   @scenecollection.deletecache()
                   @@scenes.deletecache()
                   edata = d["eventData"]
@@ -273,7 +259,7 @@ module OBS
                   # FIXME
                 when "InputCreated"
                   edata = d["eventData"]
-                  @@inputs[edata["inputName"].to_s] = OBS::Input.new( @@reqchan, edata )
+                  @@inputs[edata["inputName"].to_s] = OBS::Input.new( self, edata )
                 when "InputNameChanged"
                   edata = d["eventData"]
                   @@inputs.renamecache( edata["oldInputName"].as_s, edata["inputName"].as_s )
@@ -351,11 +337,11 @@ module OBS
                 if d["requestStatus"]["result"].as_bool == false
                   STDERR.puts( "ERROR: #{d["requestType"].as_s} #{d["requestStatus"]["code"].as_i64}: #{d["requestStatus"]["comment"].as_s}" )
                 end
-                if openrequests[ d["requestId"].as_s ]?
-                  channel = openrequests[ d["requestId"].as_s ]
+                if @@openrequests[ d["requestId"].as_s ]?
+                  channel = @@openrequests[ d["requestId"].as_s ]
                   channel.send( d )
                   channel.close
-                  openrequests.delete( d["requestId"].as_s )
+                  @@openrequests.delete( d["requestId"].as_s )
                 else
                   STDERR.puts( "PUBSUBREAD THREAD: no response channel known for requestId: #{ d["requestId"].as_s }" )
                 end
@@ -363,7 +349,7 @@ module OBS
               Fiber.yield
             end
           end
-          @obs_pubsub.run
+          @@obs_pubsub.not_nil!.run
         rescue ex : Socket::ConnectError | IO::Error
           if @connecterror < 3
             STDERR.print( ex.message )
@@ -371,52 +357,56 @@ module OBS
             ( @connecterror == 3 ) && ( @connecterror += 1 ) && STDERR.print( ". Suppressing further repeat errors." )
             STDERR.print( "\n" )
           end
+          unless retry
+            @@negotiationdelayqueue.times do
+              @@negotiationdelay.send( ex )
+              @@negotiationdelayqueue -= 1
+            end
+          end
         rescue ex
           STDERR.print( ex.pretty_inspect )
           STDERR.print( ex.backtrace?.pretty_inspect )
         ensure
-          @closed || @obs_pubsub.close
+          @closed || @@obs_pubsub && @@obs_pubsub.not_nil!.close
           @negotiated = false
-          if @@writer
-            unless @@writer.not_nil!.dead? # wtf is the type checker smoking that Nil would be an option here?
-              @@reqchan.send( { nil, "break" } ) # ask writer fiber to terminate
+          unless retry
+            @@negotiationdelayqueue.times do
+              @@negotiationdelay.send( nil )
+              @@negotiationdelayqueue -= 1
             end
-            @@writer = nil
           end
           sleep 10
           # invalidate state on everything
-          @scenecollection = OBS::SceneCollection.new( @@reqchan, @@inputs )
-          @@inputs = OBS::Inputs.new( @@reqchan )
-          @@scenes = OBS::SceneList.new( @@reqchan, @@inputs )
-          @sources = OBS::Sources.new( @@reqchan, @@inputs, @@scenes )
-          @outputs = OBS::Outputs.new( @@reqchan )
-          @video = OBS::VideoSettings.new( @@reqchan )
-          @stats = OBS::Stats.new( @@reqchan )
+          @scenecollection = OBS::SceneCollection.new( self )
+          @@inputs = OBS::Inputs.new( self )
+          @@scenes = OBS::SceneList.new( self, @@inputs )
+          @sources = OBS::Sources.new( self, @@inputs, @@scenes )
+          @outputs = OBS::Outputs.new( self )
+          @video = OBS::VideoSettings.new( self )
+          @stats = OBS::Stats.new( self )
           retry || ( @shutdown = true )
         end
       end
     end
     def close
       @shutdown = true
-      # Does it matter if the writer thread is even running, here?
-      @@reqchan.send( { nil, "break" } ) # ask writer fiber to terminate
-      @obs_pubsub && @obs_pubsub.close
+      @@obs_pubsub && @@obs_pubsub.close
     end
   end
 
   class Outputs
     @outputs = Hash( String, OBS::Output ).new
-    def initialize( @reqchan : Channel( Tuple( Channel( JSON::Any ) | Nil, String ) ) )
+    def initialize( @websocket : OBS::WebSocket )
     end
     def populate
       reschan = Channel( JSON::Any ).new
-      @reqchan.send( { reschan, OBS.req( "GetOutputList", UUID.random.to_s ) } )
+      @websocket.send( reschan, OBS.req( "GetOutputList", UUID.random.to_s ) )
       d = reschan.receive
       if ( rdata = d["responseData"]? )
         STDERR.print( "OUTPUTS THREAD: " )
         STDERR.print( rdata.pretty_inspect )
         rdata["outputs"].as_a.each do | output |
-          @outputs[output["outputName"].as_s] = OBS::Output.new( @reqchan, output )
+          @outputs[output["outputName"].as_s] = OBS::Output.new( @websocket, output )
         end
       else
         raise ResponseDataMissingException.new( "#{d["requestType"].as_s} #{d["requestStatus"]["code"].as_i64}: #{d["requestStatus"]["comment"].as_s}" )
@@ -436,9 +426,9 @@ module OBS
     getter status : OBS::OutputStatus
     getter flags : Hash( String, Bool )
     getter state : Hash( String, String | Bool | Int64 | Float64 | Hash( String, Bool ) )
-    def initialize( @reqchan : Channel( Tuple( Channel( JSON::Any ) | Nil, String ) ), json : JSON::Any )
+    def initialize( @websocket : OBS::WebSocket, json : JSON::Any )
       @flags = Hash( String, Bool ).from_json(json["outputFlags"].to_json)
-      @status = OBS::OutputStatus.new( @reqchan, json["outputName"].as_s )
+      @status = OBS::OutputStatus.new( @websocket, json["outputName"].as_s )
       @state = Hash(String, String | Bool | Int64 | Float64 | Hash( String, Bool ) ).from_json(json.to_json)
     end
   end
@@ -447,11 +437,11 @@ module OBS
     @age = Int64.new( 0 )
     @status = Hash(String, String | Bool | Int64 | Float64 ).new
     getter name : String
-    def initialize( @reqchan : Channel( Tuple( Channel( JSON::Any ) | Nil, String ) ), @name : String )
+    def initialize( @websocket : OBS::WebSocket, @name : String )
     end
     def populate
       reschan = Channel( JSON::Any ).new
-      @reqchan.send( { reschan, OBS.req( "GetOutputStatus", UUID.random.to_s, JSON.parse( { "outputName" => @name }.to_json ) ) } )
+      @websocket.send( reschan, OBS.req( "GetOutputStatus", UUID.random.to_s, JSON.parse( { "outputName" => @name }.to_json ) ) )
       d = reschan.receive
       if ( rdata = d["responseData"]? )
         @age = Time.utc.to_unix
@@ -474,11 +464,11 @@ module OBS
 
   class VideoSettings
     @settings = Hash(String, String | Bool | Int64 | Float64 ).new
-    def initialize( @reqchan : Channel( Tuple( Channel( JSON::Any ) | Nil, String ) ) )
+    def initialize( @websocket : OBS::WebSocket )
     end
     def populate
       reschan = Channel( JSON::Any ).new
-      @reqchan.send( { reschan, OBS.req( "GetVideoSettings", UUID.random.to_s ) } )
+      @websocket.send( reschan, OBS.req( "GetVideoSettings", UUID.random.to_s ) )
       d = reschan.receive
       if ( rdata = d["responseData"]? )
         @settings = Hash(String, String | Bool | Int64 | Float64 ).from_json(rdata.to_json)
@@ -503,11 +493,11 @@ module OBS
   class Stats
     @stats = Hash(String, String | Bool | Int64 | Float64 ).new
     @age = Int64.new( 0 )
-    def initialize( @reqchan : Channel( Tuple( Channel( JSON::Any ) | Nil, String ) ) )
+    def initialize( @websocket : OBS::WebSocket )
     end
     def populate
       reschan = Channel( JSON::Any ).new
-      @reqchan.send( { reschan, OBS.req( "GetStats", UUID.random.to_s ) } )
+      @websocket.send( reschan, OBS.req( "GetStats", UUID.random.to_s ) )
       d = reschan.receive
       if ( rdata = d["responseData"]? )
         @stats = Hash(String, String | Bool | Int64 | Float64 ).from_json(rdata.to_json)
@@ -528,19 +518,20 @@ module OBS
   end
 
   class SceneCollection # Note: distinct from SceneList
+    @inputs : OBS::Inputs
     @current = String.new
     @list = Hash( String, OBS::SceneList ).new
-    @reqchan = Channel( Tuple( Channel( JSON::Any ) | Nil, String ) ).new
-    def initialize( @reqchan : Channel( Tuple( Channel( JSON::Any ) | Nil, String ) ), @inputs : OBS::Inputs )
+    def initialize( @websocket : OBS::WebSocket )
+      @inputs = @websocket.inputs
     end
     def populate
       reschan = Channel( JSON::Any ).new
-      @reqchan.send( { reschan, OBS.req( "GetSceneCollectionList", UUID.random.to_s ) } )
+      @websocket.send( reschan, OBS.req( "GetSceneCollectionList", UUID.random.to_s ) )
       d = reschan.receive
       if ( rdata = d["responseData"]? )
         @current = rdata["currentSceneCollectionName"].as_s
         rdata["sceneCollections"].as_a.each{ | scenecollection |
-          @list[scenecollection.as_s] = OBS::SceneList.new( @reqchan, @inputs )
+          @list[scenecollection.as_s] = OBS::SceneList.new( @websocket, @inputs )
         }
       else
         raise ResponseDataMissingException.new( "#{d["requestType"].as_s} #{d["requestStatus"]["code"].as_i64}: #{d["requestStatus"]["comment"].as_s}" )
@@ -571,11 +562,11 @@ module OBS
     @scenes = Hash(String, OBS::Scene).new
     @program = String.new
     @preview = String.new
-    def initialize( @reqchan : Channel( Tuple( Channel( JSON::Any ) | Nil, String ) ), @inputs : OBS::Inputs )
+    def initialize( @websocket : OBS::WebSocket, @inputs : OBS::Inputs )
     end
     def populate
       reschan = Channel( JSON::Any ).new
-      @reqchan.send( { reschan, OBS.req( "GetSceneList", UUID.random.to_s ) } )
+      @websocket.send( reschan, OBS.req( "GetSceneList", UUID.random.to_s ) )
       d = reschan.receive
       if ( rdata = d["responseData"]? )
         @program = rdata["currentProgramSceneName"].as_s
@@ -583,7 +574,7 @@ module OBS
         rdata["scenes"].as_a.each_with_index{ | scene, i |
           name = scene["sceneName"].as_s
           @scenes_by_index.push(name)
-          @scenes[name] = OBS::Scene.new( @reqchan, name, self )
+          @scenes[name] = OBS::Scene.new( @websocket, name, self )
         }
       else
         raise ResponseDataMissingException.new( "#{d["requestType"].as_s} #{d["requestStatus"]["code"].as_i64}: #{d["requestStatus"]["comment"].as_s}" )
@@ -649,19 +640,19 @@ module OBS
     getter name     : String
     getter parent   : OBS::SceneList # needed for reverse metasceneitem lookup
     @age            = Int64.new( 0 )
-    def initialize( @reqchan : Channel( Tuple( Channel( JSON::Any ) | Nil, String ) ), @name : String, @parent : OBS::SceneList )
-      @filters  = OBS::SourceFilters.new( @reqchan, @name )
+    def initialize( @websocket : OBS::WebSocket, @name : String, @parent : OBS::SceneList )
+      @filters  = OBS::SourceFilters.new( @websocket, @name )
     end
     def populate
       reschan = Channel( JSON::Any ).new
-      @reqchan.send( { reschan, OBS.req( "GetSceneItemList", UUID.random.to_s, JSON.parse( { "sceneName" => @name }.to_json ) ) } )
+      @websocket.send( reschan, OBS.req( "GetSceneItemList", UUID.random.to_s, JSON.parse( { "sceneName" => @name }.to_json ) ) )
       d = reschan.receive
       if ( rdata = d["responseData"]? )
         # handle "sourceType": "OBS_SOURCE_TYPE_SCENE"
         rdata["sceneItems"].as_a.each_with_index{ | item, i |
           sname = item["sourceName"].as_s
           siid   = item["sceneItemId"].as_i64
-          @items_by_id[siid] = OBS::SceneItem.new( @reqchan, @name, item )
+          @items_by_id[siid] = OBS::SceneItem.new( @websocket, @name, item )
           @items_by_index.push( siid )
           @items_by_name[sname] = siid
           if item["sourceType"].as_s == "OBS_SOURCE_TYPE_SCENE"
@@ -748,12 +739,12 @@ module OBS
     end
     def program!
       reschan = Channel( JSON::Any ).new
-      @reqchan.send( { reschan, OBS.req( "SetCurrentProgramScene", UUID.random.to_s, JSON.parse( { "sceneName" => @name }.to_json ) ) } )
+      @websocket.send( reschan, OBS.req( "SetCurrentProgramScene", UUID.random.to_s, JSON.parse( { "sceneName" => @name }.to_json ) ) )
       reschan.receive # we get weird behavior on transitions unless we block here
     end
     def preview!
       reschan = Channel( JSON::Any ).new
-      @reqchan.send( { reschan, OBS.req( "SetCurrentPreviewScene", UUID.random.to_s, JSON.parse( { "sceneName" => @name }.to_json ) ) } )
+      @websocket.send( reschan, OBS.req( "SetCurrentPreviewScene", UUID.random.to_s, JSON.parse( { "sceneName" => @name }.to_json ) ) )
       reschan.receive # we get weird behavior on transitions unless we block here
     end
     def createinput( iname : String, kind : String, settings : OBS::InputSettings | Hash( String, String | Bool | Int64 | Float64 ) | Nil = nil, enabled : Bool = true )
@@ -769,10 +760,10 @@ module OBS
       end
 
       reschan = Channel( JSON::Any ).new
-      @reqchan.send( { reschan, OBS.req( "CreateInput", UUID.random.to_s, JSON.parse(reqdata.to_json) ) } )
+      @websocket.send( reschan, OBS.req( "CreateInput", UUID.random.to_s, JSON.parse(reqdata.to_json) ) )
       d = reschan.receive
       # maybe block until the event comes in?
-      # will have to depend on the EventSubscription state
+      # will have to depend on the @@eventsubscription state
       if d["requestStatus"]["result"].as_bool # success?
         # "responseData": { "sceneItemId": 219 }
         # Is this ever useful? It's a lot more requests than to just redo the entire GetSceneItemList
@@ -807,7 +798,7 @@ module OBS
     getter blendmode : ( String | Nil )
     getter locked    : Bool
     getter transform : OBS::SceneItemTransform
-    def initialize( @reqchan : Channel( Tuple( Channel( JSON::Any ) | Nil, String ) ), @scenename : String, json : JSON::Any )
+    def initialize( @websocket : OBS::WebSocket, @scenename : String, json : JSON::Any )
       @id      = json["sceneItemId"].as_i64
       @name    = json["sourceName"].as_s
       @kind    = json["inputKind"]?.as_s?
@@ -815,7 +806,7 @@ module OBS
       @locked  = json["sceneItemLocked"].as_bool
       @enabled = json["sceneItemEnabled"].as_bool
       @blendmode = json["sceneItemBlendmode"]?.as_s?
-      @transform = OBS::SceneItemTransform.new( @reqchan, json["sceneItemTransform"], @name, @id )
+      @transform = OBS::SceneItemTransform.new( @websocket, json["sceneItemTransform"], @name, @id )
     end
     def enabledcache( v : Bool )
       @enabled = v
@@ -832,12 +823,12 @@ module OBS
     end
     def enable!
       reschan = Channel( JSON::Any ).new
-      @reqchan.send( { reschan, OBS.req( "SetSceneItemEnabled", UUID.random.to_s, JSON.parse( { "sceneName" => @scenename, "sceneItemId" => @id, "sceneItemEnabled" => true }.to_json ) ) } )
+      @websocket.send( reschan, OBS.req( "SetSceneItemEnabled", UUID.random.to_s, JSON.parse( { "sceneName" => @scenename, "sceneItemId" => @id, "sceneItemEnabled" => true }.to_json ) ) )
       d = reschan.receive # block until response
     end
     def disable!
       reschan = Channel( JSON::Any ).new
-      @reqchan.send( { reschan, OBS.req( "SetSceneItemEnabled", UUID.random.to_s, JSON.parse( { "sceneName" => @scenename, "sceneItemId" => @id, "sceneItemEnabled" => false }.to_json ) ) } )
+      @websocket.send( reschan, OBS.req( "SetSceneItemEnabled", UUID.random.to_s, JSON.parse( { "sceneName" => @scenename, "sceneItemId" => @id, "sceneItemEnabled" => false }.to_json ) ) )
       d = reschan.receive # block until response
     end
     def transform( newtransform : OBS::SceneItemTransform | Hash( String, String | Bool | Int64 | Float64 ) )
@@ -846,19 +837,19 @@ module OBS
       reqdata["sceneName"] = @scenename
       reqdata["sceneItemTransform"] = Hash( String, String | Bool | Int64 | Float64).new.merge( newtransform.to_h )
       reschan = Channel( JSON::Any ).new
-      @reqchan.send( { reschan, OBS.req( "SetSceneItemTransform", UUID.random.to_s, JSON.parse(reqdata.to_json) ) } )
+      @websocket.send( reschan, OBS.req( "SetSceneItemTransform", UUID.random.to_s, JSON.parse(reqdata.to_json) ) )
       d = reschan.receive
     end
   end
 
   class SceneItemTransform
     @sceneitemtransform = Hash(String, String | Bool | Int64 | Float64 ).new
-    def initialize( @reqchan : Channel( Tuple( Channel( JSON::Any ) | Nil, String ) ), json : JSON::Any, @sceneitemname : String, @sceneitemid : Int64 )
+    def initialize( @websocket : OBS::WebSocket, json : JSON::Any, @sceneitemname : String, @sceneitemid : Int64 )
       @sceneitemtransform = Hash(String, String | Bool | Int64 | Float64 ).from_json(json.to_json)
     end
     def populate
       reschan = Channel( JSON::Any ).new
-      @reqchan.send( { reschan, OBS.req( "GetSceneItemTransform", UUID.random.to_s, JSON.parse( { "sceneName" => @sceneitemname, "sceneItemId" => @sceneitemid }.to_json ) ) } )
+      @websocket.send( reschan, OBS.req( "GetSceneItemTransform", UUID.random.to_s, JSON.parse( { "sceneName" => @sceneitemname, "sceneItemId" => @sceneitemid }.to_json ) ) )
       d = reschan.receive
       if ( rdata = d["responseData"]? )
         @sceneitemtransform = Hash(String, String | Bool | Int64 | Float64 ).from_json(rdata["sceneItemTransform"].to_json)
@@ -883,7 +874,7 @@ module OBS
     getter scenes : OBS::SceneList
     getter inputs : OBS::Inputs
     @last_known_real_transform = Hash( String, OBS::SceneItemTransform ).new
-    def initialize( @reqchan : Channel( Tuple( Channel( JSON::Any ) | Nil, String ) ), @inputs : OBS::Inputs, @scenes : OBS::SceneList )
+    def initialize( @websocket : OBS::WebSocket, @inputs : OBS::Inputs, @scenes : OBS::SceneList )
     end
     def []?( index : String ) : OBS::Scene | OBS::Input | Nil
       if @inputs[index]?
@@ -898,7 +889,7 @@ module OBS
       return @inputs.to_h.merge( @scenes.to_h )
     end
     def last_known_real_transform_add( sourcename : String, transform : JSON::Any )
-      @last_known_real_transform[ sourcename ] = OBS::SceneItemTransform.new( @reqchan, transform, sourcename, 0.to_i64 )
+      @last_known_real_transform[ sourcename ] = OBS::SceneItemTransform.new( @websocket, transform, sourcename, 0.to_i64 )
     end
     def last_known_real_transform( sourcename : String ) : OBS::SceneItemTransform
       @last_known_real_transform[ sourcename ]
@@ -910,15 +901,15 @@ module OBS
 
   class Inputs
     @inputs = Hash( String, OBS::Input ).new
-    def initialize( @reqchan : Channel( Tuple( Channel( JSON::Any ) | Nil, String ) ) )
+    def initialize( @websocket : OBS::WebSocket )
     end
     def populate
       reschan = Channel( JSON::Any ).new
-      @reqchan.send( { reschan, OBS.req( "GetInputList", UUID.random.to_s) } )
+      @websocket.send( reschan, OBS.req( "GetInputList", UUID.random.to_s) )
       d = reschan.receive
       if ( rdata = d["responseData"]? )
         rdata["inputs"].as_a.each{ | input |
-          @inputs[input["inputName"].as_s] = OBS::Input.new( @reqchan, input )
+          @inputs[input["inputName"].as_s] = OBS::Input.new( @websocket, input )
         }
       else
         raise ResponseDataMissingException.new( "#{d["requestType"].as_s} #{d["requestStatus"]["code"].as_i64}: #{d["requestStatus"]["comment"].as_s}" )
@@ -957,15 +948,15 @@ module OBS
     getter name : String
     getter settings : OBS::InputSettings
     getter filters  : OBS::SourceFilters
-    def initialize( @reqchan : Channel( Tuple( Channel( JSON::Any ) | Nil, String ) ), json : JSON::Any )
+    def initialize( @websocket : OBS::WebSocket, json : JSON::Any )
       @kind = json["inputKind"].as_s
       @name = json["inputName"].as_s
-      @settings = OBS::InputSettings.new( @reqchan, @name )
-      @filters  = OBS::SourceFilters.new( @reqchan, @name )
+      @settings = OBS::InputSettings.new( @websocket, @name )
+      @filters  = OBS::SourceFilters.new( @websocket, @name )
     end
     def delete!( )
       reschan = Channel( JSON::Any ).new
-      @reqchan.send( { reschan, OBS.req( "RemoveInput", UUID.random.to_s, JSON.parse({ "inputName" => @name }.to_json) ) } )
+      @websocket.send( reschan, OBS.req( "RemoveInput", UUID.random.to_s, JSON.parse({ "inputName" => @name }.to_json) ) )
       d = reschan.receive
     end
     def namecache( name : String )
@@ -982,13 +973,13 @@ module OBS
     @inputsettings = Hash(String, String | Bool | Int64 | Float64 ).new
     @inputkind = String.new
     getter name : String
-    def initialize( @reqchan : Channel( Tuple( Channel( JSON::Any ) | Nil, String ) ), @name : String )
+    def initialize( @websocket : OBS::WebSocket, @name : String )
     end
-    def initialize( @reqchan : Channel( Tuple( Channel( JSON::Any ) | Nil, String ) ), @name : String, @inputkind : String, @inputsettings : Hash( String, String | Bool | Int64 | Float64 ) )
+    def initialize( @websocket : OBS::WebSocket, @name : String, @inputkind : String, @inputsettings : Hash( String, String | Bool | Int64 | Float64 ) )
     end
     def populate
       reschan = Channel( JSON::Any ).new
-      @reqchan.send( { reschan, OBS.req( "GetInputSettings", UUID.random.to_s, JSON.parse( { "inputName" => @name }.to_json ) ) } )
+      @websocket.send( reschan, OBS.req( "GetInputSettings", UUID.random.to_s, JSON.parse( { "inputName" => @name }.to_json ) ) )
       d = reschan.receive
       if ( rdata = d["responseData"]? )
         @inputsettings = Hash(String, String | Bool | Int64 | Float64 ).from_json(rdata["inputSettings"].to_json)
@@ -1018,17 +1009,17 @@ module OBS
     @filters_by_name = Hash( String, OBS::SourceFilter ).new
     @filters_by_index = Array( String ).new
     getter sourcename : String
-    def initialize( @reqchan : Channel( Tuple( Channel( JSON::Any ) | Nil, String ) ), @sourcename : String )
+    def initialize( @websocket : OBS::WebSocket, @sourcename : String )
     end
     def populate
       reschan = Channel( JSON::Any ).new
-      @reqchan.send( { reschan, OBS.req( "GetSourceFilterList", UUID.random.to_s, JSON.parse( { "sourceName" => @sourcename }.to_json ) ) } )
+      @websocket.send( reschan, OBS.req( "GetSourceFilterList", UUID.random.to_s, JSON.parse( { "sourceName" => @sourcename }.to_json ) ) )
       d = reschan.receive
       if ( rdata = d["responseData"]? )
         rdata["filters"].as_a.each do | filter |
           name = filter["filterName"].as_s
           @filters_by_index.push( name )
-          @filters_by_name[name] = OBS::SourceFilter.new( @reqchan, @sourcename, filter )
+          @filters_by_name[name] = OBS::SourceFilter.new( @websocket, @sourcename, filter )
         end
       else
         raise ResponseDataMissingException.new( "#{d["requestType"].as_s} #{d["requestStatus"]["code"].as_i64}: #{d["requestStatus"]["comment"].as_s}" )
@@ -1059,11 +1050,11 @@ module OBS
     getter enabled    : Bool
     getter settings   : OBS::SourceFilterSettings
     getter sourcename : String
-    def initialize( @reqchan : Channel( Tuple( Channel( JSON::Any ) | Nil, String ) ), @sourcename, json : JSON::Any )
+    def initialize( @websocket : OBS::WebSocket, @sourcename, json : JSON::Any )
       @kind = json["filterKind"].as_s
       @name = json["filterName"].as_s
       @enabled = json["filterEnabled"].as_bool
-      @settings = OBS::SourceFilterSettings.new( @reqchan, @name, json["filterSettings"] )
+      @settings = OBS::SourceFilterSettings.new( @websocket, @name, json["filterSettings"] )
     end
     def toggle!
       if @enabled
@@ -1074,13 +1065,13 @@ module OBS
     end
     def enable!
       reschan = Channel( JSON::Any ).new
-      @reqchan.send( { reschan, OBS.req( "SetSourceFilterEnabled", UUID.random.to_s, JSON.parse( { "sourceName" => @sourcename, "filterName" => @name, "filterEnabled" => true }.to_json ) ) } )
+      @websocket.send( reschan, OBS.req( "SetSourceFilterEnabled", UUID.random.to_s, JSON.parse( { "sourceName" => @sourcename, "filterName" => @name, "filterEnabled" => true }.to_json ) ) )
       d = reschan.receive
       return d["requestStatus"]["result"].as_bool
     end
     def disable!
       reschan = Channel( JSON::Any ).new
-      @reqchan.send( { reschan, OBS.req( "SetSourceFilterEnabled", UUID.random.to_s, JSON.parse( { "sourceName" => @sourcename, "filterName" => @name, "filterEnabled" => false }.to_json ) ) } )
+      @websocket.send( reschan, OBS.req( "SetSourceFilterEnabled", UUID.random.to_s, JSON.parse( { "sourceName" => @sourcename, "filterName" => @name, "filterEnabled" => false }.to_json ) ) )
       d = reschan.receive
       return d["requestStatus"]["result"].as_bool
     end
@@ -1095,7 +1086,7 @@ module OBS
   class SourceFilterSettings
     getter settings : Hash(String, String | Bool | Int64 | Float64 )
     getter sourcename : String
-    def initialize( @reqchan : Channel( Tuple( Channel( JSON::Any ) | Nil, String ) ), @sourcename, json : JSON::Any )
+    def initialize( @websocket : OBS::WebSocket, @sourcename, json : JSON::Any )
       @settings = Hash(String, String | Bool | Int64 | Float64 ).from_json(json.to_json)
     end
     # do we need to populate() this somehow?