!stream: set callbacks to Connection, not Stream (for performance)

5 files changed, 23 insertions, 29 deletions

diff --git a/lib/plum/event_emitter.rb b/lib/plum/event_emitter.rb
index feebfa1..3aa8e87 100644
--- a/lib/plum/event_emitter.rb
+++ b/lib/plum/event_emitter.rb
@@ -7,11 +7,11 @@ module Plum
       callbacks[name] << blk
     end
 
-    private
     def callback(name, *args)
       callbacks[name].each {|cb| cb.call(*args) }
     end
 
+    private
     def callbacks
       @callbacks ||= Hash.new {|hash, key| hash[key] = [] }
     end
diff --git a/lib/plum/stream.rb b/lib/plum/stream.rb
index 007d352..7987532 100644
--- a/lib/plum/stream.rb
+++ b/lib/plum/stream.rb
@@ -2,7 +2,6 @@ using Plum::BinaryString
 
 module Plum
   class Stream
-    include EventEmitter
     include FlowControl
     include StreamUtils
 
@@ -53,7 +52,7 @@ module Plum
         # MUST ignore unknown frame
       end
     rescue StreamError => e
-      callback(:stream_error, e)
+      connection.callback(:stream_error, self, e)
       close(e.http2_error_type)
     end
 
@@ -96,7 +95,7 @@ module Plum
     end
 
     def receive_end_stream
-      callback(:end_stream)
+      connection.callback(:end_stream, self)
       @state = :half_closed_remote
     end
 
@@ -114,7 +113,7 @@ module Plum
       else
         body = frame.payload
       end
-      callback(:data, body)
+      connection.callback(:data, self, body)
 
       receive_end_stream if frame.end_stream?
     end
@@ -153,7 +152,7 @@ module Plum
         raise ConnectionError.new(:compression_error, e)
       end
 
-      callback(:headers, decoded_headers)
+      connection.callback(:headers, self, decoded_headers)
 
       receive_end_stream if first.end_stream?
     end
@@ -168,7 +167,7 @@ module Plum
       end
 
       @state = :open
-      callback(:open)
+      connection.callback(:open, self)
 
       if frame.end_headers?
         receive_complete_headers([frame])
@@ -210,7 +209,7 @@ module Plum
         raise ConnectionError.new(:protocol_error)
       end
 
-      callback(:rst_stream, frame)
+      connection.callback(:rst_stream, self, frame)
       @state = :closed # MUST NOT send RST_STREAM
     end
   end
diff --git a/test/plum/stream/test_handle_frame.rb b/test/plum/stream/test_handle_frame.rb
index ee9a394..6a062cf 100644
--- a/test/plum/stream/test_handle_frame.rb
+++ b/test/plum/stream/test_handle_frame.rb
@@ -8,7 +8,7 @@ class StreamHandleFrameTest < Minitest::Test
     payload = "ABC" * 5
     open_new_stream(state: :open) {|stream|
       data = nil
-      stream.on(:data) {|_data| data = _data }
+      stream.connection.on(:data) {|_, _data| data = _data }
       stream.receive_frame(Frame.new(type: :data, stream_id: stream.id,
                                      flags: [], payload: payload))
       assert_equal(payload, data)
@@ -19,7 +19,7 @@ class StreamHandleFrameTest < Minitest::Test
     payload = "ABC" * 5
     open_new_stream(state: :open) {|stream|
       data = nil
-      stream.on(:data) {|_data| data = _data }
+      stream.connection.on(:data) {|_, _data| data = _data }
       stream.receive_frame(Frame.new(type: :data, stream_id: stream.id,
                                      flags: [:padded], payload: "".push_uint8(6).push(payload).push("\x00"*6)))
       assert_equal(payload, data)
@@ -59,7 +59,7 @@ class StreamHandleFrameTest < Minitest::Test
   def test_stream_handle_headers_single
     open_new_stream {|stream|
       headers = nil
-      stream.on(:headers) {|_headers|
+      stream.connection.on(:headers) {|_, _headers|
         headers = _headers
       }
       stream.receive_frame(Frame.new(type: :headers,
@@ -75,7 +75,7 @@ class StreamHandleFrameTest < Minitest::Test
     open_new_stream {|stream|
       payload = HPACK::Encoder.new(0).encode([[":path", "/"]])
       headers = nil
-      stream.on(:headers) {|_headers|
+      stream.connection.on(:headers) {|_, _headers|
         headers = _headers
       }
       stream.receive_frame(Frame.new(type: :headers,
@@ -96,7 +96,7 @@ class StreamHandleFrameTest < Minitest::Test
     open_new_stream {|stream|
       payload = HPACK::Encoder.new(0).encode([[":path", "/"]])
       headers = nil
-      stream.on(:headers) {|_headers|
+      stream.connection.on(:headers) {|_, _headers|
         headers = _headers
       }
       stream.receive_frame(Frame.new(type: :headers,
@@ -156,7 +156,7 @@ class StreamHandleFrameTest < Minitest::Test
       stream = open_new_stream(con)
 
       headers = nil
-      stream.on(:headers) {|_headers| headers = _headers }
+      stream.connection.on(:headers) {|_, _headers| headers = _headers }
       header_block = HPACK::Encoder.new(0).encode([[":path", "/"]])
       payload = "".push_uint32((1 << 31) | parent.id)
                   .push_uint8(50)
diff --git a/test/plum/test_http_connection.rb b/test/plum/test_http_connection.rb
index fc451d0..ea72a55 100644
--- a/test/plum/test_http_connection.rb
+++ b/test/plum/test_http_connection.rb
@@ -27,9 +27,7 @@ class HTTPConnectionNegotiationTest < Minitest::Test
     io = StringIO.new
     con = HTTPConnection.new(io)
     heads = nil
-    con.on(:stream) {|stream|
-      stream.on(:headers) {|_h| heads = _h.to_h }
-    }
+    con.on(:headers) {|_, _h| heads = _h.to_h }
     req = "GET / HTTP/1.1\r\n" <<
           "Host: rhe.jp\r\n" <<
           "User-Agent: nya\r\n" <<
diff --git a/test/plum/test_https_connection.rb b/test/plum/test_https_connection.rb
index 34679bc..b08f7be 100644
--- a/test/plum/test_https_connection.rb
+++ b/test/plum/test_https_connection.rb
@@ -64,24 +64,21 @@ class HTTPSConnectionNegotiationTest < Minitest::Test
     client_thread = Thread.new {
       sock = TCPSocket.new("127.0.0.1", LISTEN_PORT)
       begin
-        Timeout.timeout(3) {
-          ctx = OpenSSL::SSL::SSLContext.new.tap {|ctx|
-            ctx.alpn_protocols = ["h2"]
-            ctx.ciphers = "AES256-GCM-SHA384"
-          }
-          ssl = OpenSSL::SSL::SSLSocket.new(sock, ctx)
-          ssl.connect
-          ssl.write Connection::CLIENT_CONNECTION_PREFACE
-          ssl.write Frame.settings.assemble
+        ctx = OpenSSL::SSL::SSLContext.new.tap {|ctx|
+          ctx.alpn_protocols = ["h2"]
+          ctx.ciphers = "AES256-GCM-SHA384"
         }
-      rescue Timeout::Error
-        flunk "client timeout"
+        ssl = OpenSSL::SSL::SSLSocket.new(sock, ctx)
+        ssl.connect
+        ssl.write Connection::CLIENT_CONNECTION_PREFACE
+        ssl.write Frame.settings.assemble
+        sleep
       ensure
         sock.close
       end
     }
-    client_thread.join
     server_thread.join
+    client_thread.kill
 
     flunk "test not run" unless run
   end