From 286af14b37518f1e021c3bb81f192ef52f47a237 Mon Sep 17 00:00:00 2001
From: Theo <theo@iconara.net>
Date: Sat, 8 Nov 2014 16:48:47 +0100
Subject: [PATCH 01/23] Add Ione::Stream and Ione::Stream::PushStream

---
 lib/ione.rb              |   1 +
 lib/ione/stream.rb       |  98 ++++++++++++++++++++++++++++
 spec/ione/stream_spec.rb | 137 +++++++++++++++++++++++++++++++++++++++
 3 files changed, 236 insertions(+)
 create mode 100644 lib/ione/stream.rb
 create mode 100644 spec/ione/stream_spec.rb

diff --git a/lib/ione.rb b/lib/ione.rb
index 98d0b58..46c0c53 100644
--- a/lib/ione.rb
+++ b/lib/ione.rb
@@ -4,5 +4,6 @@ module Ione
 end
 
 require 'ione/future'
+require 'ione/stream'
 require 'ione/byte_buffer'
 require 'ione/io'
diff --git a/lib/ione/stream.rb b/lib/ione/stream.rb
new file mode 100644
index 0000000..9410591
--- /dev/null
+++ b/lib/ione/stream.rb
@@ -0,0 +1,98 @@
+# encoding: utf-8
+
+module Ione
+  # @abstract Base class for streams
+  # @see Ione::Stream::PushStream
+  class Stream
+    # @private
+    def initialize
+      @listeners = []
+      @lock = Mutex.new
+    end
+
+    # @yieldparam [Object] element each element that flows through the stream
+    # @return [self] the stream itself
+    def each(&listener)
+      @lock.lock
+      listeners = @listeners.dup
+      listeners << listener
+      @listeners = listeners
+      self
+    ensure
+      @lock.unlock
+    end
+
+    private
+
+    def deliver(element)
+      @listeners.each do |listener|
+        listener.call(element) rescue nil
+      end
+      self
+    end
+
+    module StreamCombinators
+      # @return [Ione::Stream]
+      def map(&transformer)
+        TransformedStream.new(self, transformer)
+      end
+
+      # @return [Ione::Stream]
+      def select(&filter)
+        FilteredStream.new(self, filter)
+      end
+
+      # @return [Ione::Stream]
+      def aggregate(state=nil, &aggregator)
+        AggregatingStream.new(self, aggregator, state)
+      end
+    end
+
+    include StreamCombinators
+
+    class PushStream < Stream
+      module_eval do
+        # this crazyness is just to hide these declarations from Yard
+        alias_method :push, :deliver
+        public :push
+        alias_method :<<, :deliver
+        public :<<
+      end
+
+      # @!parse
+      #   # @param [Object] element
+      #   # @return [self]
+      #   def push(element); end
+      #   alias_method :<<, :push
+
+      # @return [Proc] a Proc that can be used to push elements to this stream
+      def to_proc
+        method(:push).to_proc
+      end
+    end
+
+    # @private
+    class TransformedStream < Stream
+      def initialize(upstream, transformer)
+        super()
+        upstream.each { |e| deliver(transformer.call(e)) }
+      end
+    end
+
+    # @private
+    class FilteredStream < Stream
+      def initialize(upstream, filter)
+        super()
+        upstream.each { |e| deliver(e) if filter.call(e) }
+      end
+    end
+
+    # @private
+    class AggregatingStream < PushStream
+      def initialize(upstream, aggregator, state)
+        super()
+        upstream.each { |e| state = aggregator.call(e, self, state) }
+      end
+    end
+  end
+end
diff --git a/spec/ione/stream_spec.rb b/spec/ione/stream_spec.rb
new file mode 100644
index 0000000..60b0d14
--- /dev/null
+++ b/spec/ione/stream_spec.rb
@@ -0,0 +1,137 @@
+# encoding: utf-8
+
+require 'spec_helper'
+
+
+module Ione
+  class Stream
+    describe PushStream do
+      let :stream do
+        described_class.new
+      end
+
+      describe '#push' do
+        it 'pushes an element to the stream' do
+          pushed_elements = []
+          stream.each { |e| pushed_elements << e }
+          stream.push('foo')
+          pushed_elements.should == ['foo']
+        end
+
+        it 'is aliased as #<<' do
+          pushed_elements = []
+          stream.each { |e| pushed_elements << e }
+          stream << 'foo'
+          pushed_elements.should == ['foo']
+        end
+
+        it 'returns self' do
+          stream.push('foo').should equal(stream)
+        end
+
+        it 'delivers the element to all listeners' do
+          pushed_elements1 = []
+          pushed_elements2 = []
+          pushed_elements3 = []
+          stream.each { |e| pushed_elements1 << e }
+          stream.each { |e| pushed_elements2 << e }
+          stream.each { |e| pushed_elements3 << e }
+          stream.push('foo')
+          pushed_elements1.should == ['foo']
+          pushed_elements2.should == ['foo']
+          pushed_elements3.should == ['foo']
+        end
+
+        it 'ignores errors raised by listeners' do
+          pushed_elements1 = []
+          pushed_elements2 = []
+          stream.each { |e| raise 'bork!' }
+          stream.each { |e| pushed_elements2 << e }
+          stream.push('foo')
+          pushed_elements1.should == []
+          pushed_elements2.should == ['foo']
+        end
+      end
+
+      describe '#each' do
+        it 'yields each element that is pushed to the stream' do
+          pushed_elements = []
+          stream.each { |e| pushed_elements << e }
+          stream.push('foo')
+          stream.push('bar')
+          pushed_elements.should == ['foo', 'bar']
+        end
+
+        it 'returns self' do
+          stream.each { }.should equal(stream)
+        end
+      end
+
+      describe '#to_proc' do
+        it 'returns a Proc that can be used to push elements to the stream' do
+          pushed_elements = []
+          another_stream = described_class.new
+          another_stream.each { |e| pushed_elements << e}
+          stream.each(&another_stream)
+          stream << 'foo'
+          stream << 'bar'
+          pushed_elements.should == ['foo', 'bar']
+        end
+      end
+
+      describe '#map' do
+        it 'returns a stream that will yield elements transformed by the specified block' do
+          pushed_elements = []
+          transformed_stream = stream.map { |e| e.reverse }
+          transformed_stream.each { |e| pushed_elements << e }
+          stream << 'foo'
+          stream << 'bar'
+          pushed_elements.should == ['oof', 'rab']
+        end
+      end
+
+      describe '#select' do
+        it 'returns a stream that will yield only the elements for which the specified block returns true' do
+          pushed_elements = []
+          filtered_stream = stream.select { |e| e.include?('a') }
+          filtered_stream.each { |e| pushed_elements << e }
+          stream << 'foo'
+          stream << 'bar'
+          stream << 'baz'
+          stream << 'qux'
+          pushed_elements.should == ['bar', 'baz']
+        end
+      end
+
+      describe '#aggregate' do
+        it 'returns a stream that will yield new elements produced by the specified block' do
+          pushed_elements = []
+          sum = 0
+          aggregate_stream = stream.aggregate do |e, downstream|
+            sum += e
+            downstream << sum
+          end
+          aggregate_stream.each { |e| pushed_elements << e }
+          1.upto(5) { |n| stream.push(n) }
+          pushed_elements.should == [1, 1 + 2, 1 + 2 + 3, 1 + 2 + 3 + 4, 1 + 2 + 3 + 4 + 5]
+        end
+
+        it 'passes the given argument to the first invocation of the block, and the block\'s return value on each subsequent invocation' do
+          pushed_elements = []
+          aggregate_stream = stream.aggregate('') do |e, downstream, buffer|
+            buffer << e
+            while (i = buffer.index("\n"))
+              downstream << buffer.slice!(0, i + 1)
+            end
+            buffer
+          end
+          aggregate_stream.each { |e| pushed_elements << e }
+          stream << "fo"
+          stream << "o\nbar\nba"
+          stream << "z\n"
+          pushed_elements.should == ["foo\n", "bar\n", "baz\n"]
+        end
+      end
+    end
+  end
+end

From 8d6b0d53bd06224e42693c0a5817e504088b5806 Mon Sep 17 00:00:00 2001
From: Theo <theo@iconara.net>
Date: Sat, 8 Nov 2014 17:13:43 +0100
Subject: [PATCH 02/23] Use PushStream for the data listening in *Connection

---
 lib/ione/io/base_connection.rb | 7 ++++---
 lib/ione/io/ssl_connection.rb  | 6 ++----
 2 files changed, 6 insertions(+), 7 deletions(-)

diff --git a/lib/ione/io/base_connection.rb b/lib/ione/io/base_connection.rb
index 2904011..e530a28 100644
--- a/lib/ione/io/base_connection.rb
+++ b/lib/ione/io/base_connection.rb
@@ -21,6 +21,7 @@ def initialize(host, port, unblocker)
         @lock = Mutex.new
         @write_buffer = ByteBuffer.new
         @closed_promise = Promise.new
+        @data_stream = Stream::PushStream.new
       end
 
       # Closes the connection
@@ -103,7 +104,8 @@ def writable?
       #
       # @yield [String] the new data
       def on_data(&listener)
-        @data_listener = listener
+        @data_stream.each(&listener)
+        nil
       end
 
       # Register to receive a notification when the socket is closed, both for
@@ -166,8 +168,7 @@ def flush
 
       # @private
       def read
-        new_data = @io.read_nonblock(65536)
-        @data_listener.call(new_data) if @data_listener
+        @data_stream << @io.read_nonblock(65536)
       rescue => e
         close(e)
       end
diff --git a/lib/ione/io/ssl_connection.rb b/lib/ione/io/ssl_connection.rb
index 5b89009..662b015 100644
--- a/lib/ione/io/ssl_connection.rb
+++ b/lib/ione/io/ssl_connection.rb
@@ -42,8 +42,7 @@ def to_io
         # @private
         def read
           while true
-            new_data = @io.read_nonblock(2**16)
-            @data_listener.call(new_data) if @data_listener
+            @data_stream << @io.read_nonblock(2**16)
           end
         rescue IO::WaitReadable, IO::WaitWritable
           # no more data available
@@ -55,8 +54,7 @@ def read
         def read
           read_size = 2**16
           while read_size > 0
-            new_data = @io.read_nonblock(read_size)
-            @data_listener.call(new_data) if @data_listener
+            @data_stream << @io.read_nonblock(read_size)
             read_size = @io.pending
           end
         rescue IO::WaitReadable, IO::WaitWritable

From 123245b2f01ccf666af835143de844d5a05b552c Mon Sep 17 00:00:00 2001
From: Theo <theo@iconara.net>
Date: Sat, 8 Nov 2014 17:24:38 +0100
Subject: [PATCH 03/23] Add *Connection#to_stream

A connection is a stream of the data chunks received by the socket
---
 lib/ione/io/base_connection.rb    | 23 +++++++++++++++++++++++
 spec/ione/io/connection_common.rb | 12 ++++++++++++
 2 files changed, 35 insertions(+)

diff --git a/lib/ione/io/base_connection.rb b/lib/ione/io/base_connection.rb
index e530a28..500932b 100644
--- a/lib/ione/io/base_connection.rb
+++ b/lib/ione/io/base_connection.rb
@@ -85,6 +85,29 @@ def writable?
         @writable && @state != CLOSED_STATE
       end
 
+      # Returns a stream of data chunks received by this connection
+      #
+      # It is very important that you don't do any heavy lifting in subscribers
+      # to this stream since they will be called from the IO reactor thread.
+      #
+      # @example Transforming a stream of data chunks to a stream of lines
+      #   data_chunk_stream = connection.to_stream
+      #   line_stream = data_chunk_stream.aggregate(ByteBuffer.new) do |chunk, downstream, buffer|
+      #     buffer << chunk
+      #     while (newline_index = buffer.index("\n"))
+      #       downstream << buffer.read(newline_index + 1)
+      #     end
+      #     buffer
+      #   end
+      #   line_stream.each do |line|
+      #     puts line
+      #   end
+      #
+      # @return [Ione::Stream<String>]
+      def to_stream
+        @data_stream
+      end
+
       # Register to receive notifications when new data is read from the socket.
       #
       # You should only call this method in your protocol handler constructor.
diff --git a/spec/ione/io/connection_common.rb b/spec/ione/io/connection_common.rb
index a8d25f3..4fd3b39 100644
--- a/spec/ione/io/connection_common.rb
+++ b/spec/ione/io/connection_common.rb
@@ -254,5 +254,17 @@
         end
       end
     end
+
+    describe '#to_stream' do
+      it 'returns a stream of the data chunks received by the socket' do
+        socket.should_receive(:read_nonblock).and_return('foo bar', 'baz', 'qux')
+        received_chunks = []
+        handler.to_stream.each { |data| received_chunks << data }
+        handler.read
+        handler.read
+        handler.read
+        received_chunks.should == ['foo bar', 'baz', 'qux']
+      end
+    end
   end
 end

From dc729144786e24468cd2f1e45d56374a960f6ea5 Mon Sep 17 00:00:00 2001
From: Theo <theo@iconara.net>
Date: Sat, 8 Nov 2014 17:27:35 +0100
Subject: [PATCH 04/23] Rewrite the Redis client example with streams

---
 .../redis_client/lib/ione/redis_client.rb     | 206 +++++++++---------
 1 file changed, 102 insertions(+), 104 deletions(-)

diff --git a/examples/redis_client/lib/ione/redis_client.rb b/examples/redis_client/lib/ione/redis_client.rb
index f931d46..91fc17f 100644
--- a/examples/redis_client/lib/ione/redis_client.rb
+++ b/examples/redis_client/lib/ione/redis_client.rb
@@ -13,60 +13,48 @@ def initialize(host, port)
       @host = host
       @port = port
       @reactor = Ione::Io::IoReactor.new
+      @responses = []
     end
 
     def connect
       f = @reactor.start
       f = f.flat_map { @reactor.connect(@host, @port) }
-      f = f.map { |connection| RedisProtocolHandler.new(connection) }
-      f.on_value { |protocol_handler| @protocol_handler = protocol_handler }
+      f.on_value do |connection|
+        @connection = connection
+        process_responses(connection.to_stream)
+      end
       f.map(self)
     end
 
-    def method_missing(name, *args)
-      @protocol_handler.send_request(name, *args)
-    end
-  end
-
-  class LineProtocolHandler
-    def initialize(connection)
-      @connection = connection
-      @connection.on_data(&method(:process_data))
-      @buffer = Ione::ByteBuffer.new
-      @requests = []
-    end
-
-    def on_line(&listener)
-      @line_listener = listener
-    end
-
-    def write(command_string)
-      @connection.write(command_string)
-    end
-
-    def process_data(new_data)
-      lines = []
-      @buffer << new_data
-      while (newline_index = @buffer.index("\r\n"))
-        line = @buffer.read(newline_index + 2)
-        line.chomp!
-        lines << line
+    def process_responses(byte_stream)
+      line_stream = byte_stream.aggregate(Ione::ByteBuffer.new) do |data, downstream, buffer|
+        buffer << data
+        while (newline_index = buffer.index("\r\n"))
+          line = buffer.read(newline_index + 2)
+          line.chomp!
+          downstream << line
+        end
+        buffer
       end
-      lines.each do |line|
-        @line_listener.call(line) if @line_listener
+      response_stream = line_stream.aggregate(RedisProtocol::BaseState.new) do |line, downstream, state|
+        state = state.feed_line(line)
+        if state.response?
+          downstream << [state.response, state.error?]
+        end
+        state
       end
-    end
-  end
-
-  class RedisProtocolHandler
-    def initialize(connection)
-      @line_protocol = LineProtocolHandler.new(connection)
-      @line_protocol.on_line(&method(:handle_line))
-      @responses = []
-      @state = BaseState.new(method(:handle_response))
+      response_stream.each do |response, error|
+        promise = @responses.shift
+        if error
+          promise.fail(StandardError.new(response))
+        else
+          promise.fulfill(response)
+        end
+      end
+      self
     end
 
-    def send_request(*args)
+    def method_missing(*args)
       promise = Ione::Promise.new
       @responses << promise
       request = "*#{args.size}\r\n"
@@ -74,87 +62,97 @@ def send_request(*args)
         arg_str = arg.to_s
         request << "$#{arg_str.bytesize}\r\n#{arg_str}\r\n"
       end
-      @line_protocol.write(request)
+      @connection.write(request)
       promise.future
     end
 
-    def handle_response(result, error=false)
-      promise = @responses.shift
-      if error
-        promise.fail(StandardError.new(result))
-      else
-        promise.fulfill(result)
-      end
-    end
+    module RedisProtocol
+      class State
+        attr_reader :next_state
 
-    def handle_line(line)
-      @state = @state.handle_line(line)
-    end
+        def initialize
+          @next_state = self
+        end
 
-    class State
-      def initialize(result_handler)
-        @result_handler = result_handler
-      end
+        def response?
+          false
+        end
 
-      def complete(result)
-        @result_handler.call(result)
-      end
+        def continue(next_state=self)
+          next_state
+        end
 
-      def fail(message)
-        @result_handler.call(message, true)
+        def complete(response, error=false)
+          CompleteState.new(response, error)
+        end
       end
-    end
 
-    class BulkState < State
-      def handle_line(line)
-        complete(line)
-        BaseState.new(@result_handler)
+      class BulkState < State
+        def feed_line(line)
+          complete(line)
+        end
       end
-    end
 
-    class MultiBulkState < State
-      def initialize(result_handler, expected_elements)
-        super(result_handler)
-        @expected_elements = expected_elements
-        @elements = []
-      end
+      class MultiBulkState < State
+        def initialize(expected_elements)
+          super()
+          @expected_elements = expected_elements
+          @elements = []
+        end
 
-      def handle_line(line)
-        if line.start_with?('$')
-          line.slice!(0, 1)
-          if line.to_i == -1
-            @elements << nil
+        def feed_line(line)
+          if line.start_with?('$')
+            line.slice!(0, 1)
+            if line.to_i == -1
+              @elements << nil
+            end
+          else
+            @elements << line
+          end
+          if @elements.size == @expected_elements
+            complete(@elements)
+          else
+            continue
           end
-        else
-          @elements << line
-        end
-        if @elements.size == @expected_elements
-          complete(@elements)
-          BaseState.new(@result_handler)
-        else
-          self
         end
       end
-    end
 
-    class BaseState < State
-      def handle_line(line)
-        next_state = self
-        first_char = line.slice!(0, 1)
-        case first_char
-        when '+' then complete(line)
-        when ':' then complete(line.to_i)
-        when '-' then fail(line)
-        when '$'
-          if line.to_i == -1
-            complete(nil)
+      class BaseState < State
+        def feed_line(line)
+          first_char = line.slice!(0, 1)
+          case first_char
+          when '+' then complete(line)
+          when ':' then complete(line.to_i)
+          when '-' then complete(line, true)
+          when '$'
+            if line.to_i == -1
+              complete(nil)
+            else
+              continue(BulkState.new)
+            end
+          when '*'
+            continue(MultiBulkState.new(line.to_i))
           else
-            next_state = BulkState.new(@result_handler)
+            continue
           end
-        when '*'
-          next_state = MultiBulkState.new(@result_handler, line.to_i)
         end
-        next_state
+      end
+
+      class CompleteState < BaseState
+        attr_reader :response
+
+        def initialize(response, error=false)
+          @response = response
+          @error = error
+        end
+
+        def response?
+          true
+        end
+
+        def error?
+          @error
+        end
       end
     end
   end

From d5554dc449bfc95d2464fabb56e7bfeacdeff0ae Mon Sep 17 00:00:00 2001
From: Theo <theo@iconara.net>
Date: Sat, 8 Nov 2014 17:43:47 +0100
Subject: [PATCH 05/23] Rename a variable in the Redis example

---
 examples/redis_client/lib/ione/redis_client.rb | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/examples/redis_client/lib/ione/redis_client.rb b/examples/redis_client/lib/ione/redis_client.rb
index 91fc17f..ad47c4f 100644
--- a/examples/redis_client/lib/ione/redis_client.rb
+++ b/examples/redis_client/lib/ione/redis_client.rb
@@ -94,9 +94,9 @@ def feed_line(line)
       end
 
       class MultiBulkState < State
-        def initialize(expected_elements)
+        def initialize(size)
           super()
-          @expected_elements = expected_elements
+          @size = size
           @elements = []
         end
 
@@ -109,7 +109,7 @@ def feed_line(line)
           else
             @elements << line
           end
-          if @elements.size == @expected_elements
+          if @elements.size == @size
             complete(@elements)
           else
             continue

From 9676b9e8799e7eef4ac13e72ca966b5d58852bbe Mon Sep 17 00:00:00 2001
From: Theo <theo@iconara.net>
Date: Sat, 8 Nov 2014 17:45:10 +0100
Subject: [PATCH 06/23] Write some documentation for the Redis protocol
 implementation
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

I’m pretty pleased with the state machine approach to parsing the Redis protocol. It’s a few more lines than would be strictly necessary because of all the classes, but it’s always nice with a state machine.
---
 examples/redis_client/lib/ione/redis_client.rb | 15 +++++++++++++++
 1 file changed, 15 insertions(+)

diff --git a/examples/redis_client/lib/ione/redis_client.rb b/examples/redis_client/lib/ione/redis_client.rb
index ad47c4f..3d94f43 100644
--- a/examples/redis_client/lib/ione/redis_client.rb
+++ b/examples/redis_client/lib/ione/redis_client.rb
@@ -66,6 +66,21 @@ def method_missing(*args)
       promise.future
     end
 
+    # This is an implementation of the Redis protocol as a state machine.
+    #
+    # You start with {BaseState} and call {#feed_line} with a line received
+    # from Redis. {#feed_line} will return a new state object, on which you
+    # should call {#feed_line} with the next line from Redis.
+    #
+    # The state objects returned by {#feed_line} represent either a complete or
+    # a partial response. You can check if a state represents a complete
+    # response by calling {#response?}. If this method returns true you can call
+    # {#response} to get the response (which is either a string or an array of
+    # strings). In some cases the response is an error, in which case {#error?}
+    # will return true, and {#response} will return the error message.
+    #
+    # The state that represents a complete response also works like the base
+    # state so you can feed it a line to start processing the next response.
     module RedisProtocol
       class State
         attr_reader :next_state

From 1113d877d37a573b4d0445b35073da5169d07497 Mon Sep 17 00:00:00 2001
From: Theo <theo@iconara.net>
Date: Sat, 8 Nov 2014 17:50:50 +0100
Subject: [PATCH 07/23] Refactor the Redis example to better show the three
 stream transformations

---
 examples/redis_client/lib/ione/redis_client.rb | 15 +++++++++++----
 1 file changed, 11 insertions(+), 4 deletions(-)

diff --git a/examples/redis_client/lib/ione/redis_client.rb b/examples/redis_client/lib/ione/redis_client.rb
index 3d94f43..c662a99 100644
--- a/examples/redis_client/lib/ione/redis_client.rb
+++ b/examples/redis_client/lib/ione/redis_client.rb
@@ -21,13 +21,13 @@ def connect
       f = f.flat_map { @reactor.connect(@host, @port) }
       f.on_value do |connection|
         @connection = connection
-        process_responses(connection.to_stream)
+        process_data_chunks(connection.to_stream)
       end
       f.map(self)
     end
 
-    def process_responses(byte_stream)
-      line_stream = byte_stream.aggregate(Ione::ByteBuffer.new) do |data, downstream, buffer|
+    def process_data_chunks(data_chunk_stream)
+      line_stream = data_chunk_stream.aggregate(Ione::ByteBuffer.new) do |data, downstream, buffer|
         buffer << data
         while (newline_index = buffer.index("\r\n"))
           line = buffer.read(newline_index + 2)
@@ -36,6 +36,10 @@ def process_responses(byte_stream)
         end
         buffer
       end
+      process_lines(line_stream)
+    end
+
+    def process_lines(line_stream)
       response_stream = line_stream.aggregate(RedisProtocol::BaseState.new) do |line, downstream, state|
         state = state.feed_line(line)
         if state.response?
@@ -43,6 +47,10 @@ def process_responses(byte_stream)
         end
         state
       end
+      process_responses(response_stream)
+    end
+
+    def process_responses(response_stream)
       response_stream.each do |response, error|
         promise = @responses.shift
         if error
@@ -51,7 +59,6 @@ def process_responses(byte_stream)
           promise.fulfill(response)
         end
       end
-      self
     end
 
     def method_missing(*args)

From 1803fc28bba1d4685b29d1a7c9fc3849fda13705 Mon Sep 17 00:00:00 2001
From: Theo <theo@iconara.net>
Date: Sat, 8 Nov 2014 17:52:15 +0100
Subject: [PATCH 08/23] Add RedisError to the Redis example

---
 examples/redis_client/lib/ione/redis_client.rb       | 4 +++-
 examples/redis_client/spec/ione/redis_client_spec.rb | 2 +-
 2 files changed, 4 insertions(+), 2 deletions(-)

diff --git a/examples/redis_client/lib/ione/redis_client.rb b/examples/redis_client/lib/ione/redis_client.rb
index c662a99..d9aefa4 100644
--- a/examples/redis_client/lib/ione/redis_client.rb
+++ b/examples/redis_client/lib/ione/redis_client.rb
@@ -4,6 +4,8 @@
 
 
 module Ione
+  RedisError = Class.new(StandardError)
+
   class RedisClient
     def self.connect(host, port)
       new(host, port).connect
@@ -54,7 +56,7 @@ def process_responses(response_stream)
       response_stream.each do |response, error|
         promise = @responses.shift
         if error
-          promise.fail(StandardError.new(response))
+          promise.fail(RedisError.new(response))
         else
           promise.fulfill(response)
         end
diff --git a/examples/redis_client/spec/ione/redis_client_spec.rb b/examples/redis_client/spec/ione/redis_client_spec.rb
index 39a167e..06c028f 100644
--- a/examples/redis_client/spec/ione/redis_client_spec.rb
+++ b/examples/redis_client/spec/ione/redis_client_spec.rb
@@ -46,7 +46,7 @@ module Ione
     it 'handles errors' do
       pending('Redis not running', unless: client)
       f = client.set('foo')
-      expect { f.value }.to raise_error("ERR wrong number of arguments for 'set' command")
+      expect { f.value }.to raise_error(RedisError, "ERR wrong number of arguments for 'set' command")
     end
 
     it 'handles replies with multiple elements' do

From c157eb247074e349d692134c714c8d677219dfd6 Mon Sep 17 00:00:00 2001
From: Theo <theo@iconara.net>
Date: Sat, 8 Nov 2014 17:55:04 +0100
Subject: [PATCH 09/23] Remove some unnecessary code from the Redis protocol

This is the remains of an earlier implementation.
---
 examples/redis_client/lib/ione/redis_client.rb | 7 -------
 1 file changed, 7 deletions(-)

diff --git a/examples/redis_client/lib/ione/redis_client.rb b/examples/redis_client/lib/ione/redis_client.rb
index d9aefa4..52ef82d 100644
--- a/examples/redis_client/lib/ione/redis_client.rb
+++ b/examples/redis_client/lib/ione/redis_client.rb
@@ -92,12 +92,6 @@ def method_missing(*args)
     # state so you can feed it a line to start processing the next response.
     module RedisProtocol
       class State
-        attr_reader :next_state
-
-        def initialize
-          @next_state = self
-        end
-
         def response?
           false
         end
@@ -119,7 +113,6 @@ def feed_line(line)
 
       class MultiBulkState < State
         def initialize(size)
-          super()
           @size = size
           @elements = []
         end

From 5361c2b311d0f2f45c0dc280361609bdf3e36311 Mon Sep 17 00:00:00 2001
From: Theo <theo@iconara.net>
Date: Sat, 8 Nov 2014 17:56:34 +0100
Subject: [PATCH 10/23] Mark internal methods as private in the Redis example

---
 examples/redis_client/lib/ione/redis_client.rb | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/examples/redis_client/lib/ione/redis_client.rb b/examples/redis_client/lib/ione/redis_client.rb
index 52ef82d..f4aeffd 100644
--- a/examples/redis_client/lib/ione/redis_client.rb
+++ b/examples/redis_client/lib/ione/redis_client.rb
@@ -96,6 +96,8 @@ def response?
           false
         end
 
+        private
+
         def continue(next_state=self)
           next_state
         end

From c8b858079cdf7e2683c9ddaf45b7e8bb19a16dc6 Mon Sep 17 00:00:00 2001
From: Theo <theo@iconara.net>
Date: Sat, 8 Nov 2014 17:58:45 +0100
Subject: [PATCH 11/23] Move the is-Redis-running check to a before block

---
 examples/redis_client/spec/ione/redis_client_spec.rb | 11 ++++-------
 1 file changed, 4 insertions(+), 7 deletions(-)

diff --git a/examples/redis_client/spec/ione/redis_client_spec.rb b/examples/redis_client/spec/ione/redis_client_spec.rb
index 06c028f..b61e9c7 100644
--- a/examples/redis_client/spec/ione/redis_client_spec.rb
+++ b/examples/redis_client/spec/ione/redis_client_spec.rb
@@ -13,14 +13,16 @@ module Ione
       end
     end
 
-    it 'can set a value' do
+    before do
       pending('Redis not running', unless: client)
+    end
+
+    it 'can set a value' do
       response = client.set('foo', 'bar').value
       response.should == 'OK'
     end
 
     it 'can get a value' do
-      pending('Redis not running', unless: client)
       f = client.set('foo', 'bar').flat_map do
         client.get('foo')
       end
@@ -28,7 +30,6 @@ module Ione
     end
 
     it 'can delete values' do
-      pending('Redis not running', unless: client)
       f = client.set('hello', 'world').flat_map do
         client.del('hello')
       end
@@ -36,7 +37,6 @@ module Ione
     end
 
     it 'handles nil values' do
-      pending('Redis not running', unless: client)
       f = client.del('hello').flat_map do
         client.get('hello')
       end
@@ -44,13 +44,11 @@ module Ione
     end
 
     it 'handles errors' do
-      pending('Redis not running', unless: client)
       f = client.set('foo')
       expect { f.value }.to raise_error(RedisError, "ERR wrong number of arguments for 'set' command")
     end
 
     it 'handles replies with multiple elements' do
-      pending('Redis not running', unless: client)
       f = client.del('stuff')
       f.value
       f = client.rpush('stuff', 'hello', 'world')
@@ -60,7 +58,6 @@ module Ione
     end
 
     it 'handles nil values when reading multiple elements' do
-      pending('Redis not running', unless: client)
       client.del('things')
       client.hset('things', 'hello', 'world')
       f = client.hmget('things', 'hello', 'foo')

From ca75e1af883bb97ef2095e3beb8447c5e80a76cb Mon Sep 17 00:00:00 2001
From: Theo <theo@iconara.net>
Date: Sat, 8 Nov 2014 18:01:32 +0100
Subject: [PATCH 12/23] Add a clarifying test to the Redis example
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

It’s really a naïve Redis client implementation
---
 examples/redis_client/spec/ione/redis_client_spec.rb | 5 +++++
 1 file changed, 5 insertions(+)

diff --git a/examples/redis_client/spec/ione/redis_client_spec.rb b/examples/redis_client/spec/ione/redis_client_spec.rb
index b61e9c7..4c25905 100644
--- a/examples/redis_client/spec/ione/redis_client_spec.rb
+++ b/examples/redis_client/spec/ione/redis_client_spec.rb
@@ -48,6 +48,11 @@ module Ione
       expect { f.value }.to raise_error(RedisError, "ERR wrong number of arguments for 'set' command")
     end
 
+    it 'does no validation on the command that is sent' do
+      f = client.foo('bar')
+      expect { f.value }.to raise_error(RedisError, "ERR unknown command 'foo'")
+    end
+
     it 'handles replies with multiple elements' do
       f = client.del('stuff')
       f.value

From ec1b833450bb8b02763274cd12bc90a4d5ee2427 Mon Sep 17 00:00:00 2001
From: Theo <theo@iconara.net>
Date: Sat, 8 Nov 2014 18:05:20 +0100
Subject: [PATCH 13/23] Add a little something to the readme about streams

---
 README.md | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/README.md b/README.md
index bf53ab4..6959a52 100644
--- a/README.md
+++ b/README.md
@@ -14,6 +14,10 @@ Ione is a framework for reactive programming in Ruby. It is based on the reactiv
 
 At the core of Ione is a futures API. Futures make it easy to compose asynchronous operations.
 
+## Streams
+
+Streams are a powerful abstraction for building for example composable protocol parsers.
+
 ## Evented IO
 
 A key piece of the framework is an IO reactor with which you can easily build network clients and servers.

From 097f5eef3f080bce815afafda1f3efb46d670484 Mon Sep 17 00:00:00 2001
From: Theo <theo@iconara.net>
Date: Sat, 8 Nov 2014 18:12:43 +0100
Subject: [PATCH 14/23] Add some more API docs to the stream combinators

---
 lib/ione/stream.rb | 8 ++++++++
 1 file changed, 8 insertions(+)

diff --git a/lib/ione/stream.rb b/lib/ione/stream.rb
index 9410591..2cb9cb0 100644
--- a/lib/ione/stream.rb
+++ b/lib/ione/stream.rb
@@ -32,16 +32,24 @@ def deliver(element)
     end
 
     module StreamCombinators
+      # @yieldparam [Object] element
+      # @yieldreturn [Object] the transformed element
       # @return [Ione::Stream]
       def map(&transformer)
         TransformedStream.new(self, transformer)
       end
 
+      # @yieldparam [Object] element
+      # @yieldreturn [Boolean] whether or not to pass the element downstream
       # @return [Ione::Stream]
       def select(&filter)
         FilteredStream.new(self, filter)
       end
 
+      # @yieldparam [Object] element
+      # @yieldparam [Ione::Stream::PushStream] downstream
+      # @yieldparam [Object] state
+      # @yieldreturn [Object] the next state
       # @return [Ione::Stream]
       def aggregate(state=nil, &aggregator)
         AggregatingStream.new(self, aggregator, state)

From 6a8eaefb9214ccf6db6a5da315f9d4153e554818 Mon Sep 17 00:00:00 2001
From: Theo <theo@iconara.net>
Date: Sat, 8 Nov 2014 18:44:33 +0100
Subject: [PATCH 15/23] Add Stream#take and #drop

---
 lib/ione/stream.rb       | 62 ++++++++++++++++++++++++++++++++++++++--
 spec/ione/stream_spec.rb | 34 ++++++++++++++++++++++
 2 files changed, 94 insertions(+), 2 deletions(-)

diff --git a/lib/ione/stream.rb b/lib/ione/stream.rb
index 2cb9cb0..2de4c80 100644
--- a/lib/ione/stream.rb
+++ b/lib/ione/stream.rb
@@ -12,15 +12,25 @@ def initialize
 
     # @yieldparam [Object] element each element that flows through the stream
     # @return [self] the stream itself
-    def each(&listener)
+    def subscribe(listener=nil, &block)
       @lock.lock
       listeners = @listeners.dup
-      listeners << listener
+      listeners << (listener || block)
       @listeners = listeners
       self
     ensure
       @lock.unlock
     end
+    alias_method :each, :subscribe
+
+    def unsubscribe(listener)
+      @lock.lock
+      listeners = @listeners.dup
+      listeners.delete(listener)
+      @listeners = listeners
+    ensure
+      @lock.unlock
+    end
 
     private
 
@@ -54,6 +64,22 @@ def select(&filter)
       def aggregate(state=nil, &aggregator)
         AggregatingStream.new(self, aggregator, state)
       end
+
+      # @param [Integer] n the number of elements to pass downstream before
+      #   unsubscribing
+      # @yieldparam [Object] element
+      # @return [Ione::Stream]
+      def take(n)
+        LimitedStream.new(self, n)
+      end
+
+      # @param [Integer] n the number of elements to skip before passing
+      #   elements downstream
+      # @yieldparam [Object] element
+      # @return [Ione::Stream]
+      def drop(n)
+        SkippingStream.new(self, n)
+      end
     end
 
     include StreamCombinators
@@ -102,5 +128,37 @@ def initialize(upstream, aggregator, state)
         upstream.each { |e| state = aggregator.call(e, self, state) }
       end
     end
+
+    # @private
+    class LimitedStream < Stream
+      def initialize(upstream, n)
+        super()
+        counter = 0
+        subscriber = proc do |e|
+          if counter < n
+            deliver(e)
+          else
+            upstream.unsubscribe(subscriber)
+          end
+          counter += 1
+        end
+        upstream.subscribe(subscriber)
+      end
+    end
+
+    # @private
+    class SkippingStream < Stream
+      def initialize(upstream, n)
+        super()
+        counter = 0
+        upstream.subscribe do |e|
+          if counter == n
+            deliver(e)
+          else
+            counter += 1
+          end
+        end
+      end
+    end
   end
 end
diff --git a/spec/ione/stream_spec.rb b/spec/ione/stream_spec.rb
index 60b0d14..8c59bef 100644
--- a/spec/ione/stream_spec.rb
+++ b/spec/ione/stream_spec.rb
@@ -65,6 +65,14 @@ class Stream
         it 'returns self' do
           stream.each { }.should equal(stream)
         end
+
+        it 'is aliased as #subscribe' do
+          pushed_elements = []
+          stream.subscribe { |e| pushed_elements << e }
+          stream.push('foo')
+          stream.push('bar')
+          pushed_elements.should == ['foo', 'bar']
+        end
       end
 
       describe '#to_proc' do
@@ -132,6 +140,32 @@ class Stream
           pushed_elements.should == ["foo\n", "bar\n", "baz\n"]
         end
       end
+
+      describe '#take' do
+        it 'returns a stream that will yield only the specified number of elements' do
+          pushed_elements = []
+          filtered_stream = stream.take(3)
+          filtered_stream.each { |e| pushed_elements << e }
+          stream << 'foo'
+          stream << 'bar'
+          stream << 'baz'
+          stream << 'qux'
+          pushed_elements.should == ['foo', 'bar', 'baz']
+        end
+      end
+
+      describe '#drop' do
+        it 'returns a stream that will skip the specified number of items' do
+          pushed_elements = []
+          filtered_stream = stream.drop(2)
+          filtered_stream.each { |e| pushed_elements << e }
+          stream << 'foo'
+          stream << 'bar'
+          stream << 'baz'
+          stream << 'qux'
+          pushed_elements.should == ['baz', 'qux']
+        end
+      end
     end
   end
 end

From 0cdf4c626090fa4602c61930bbd6a2f24b7fd9f1 Mon Sep 17 00:00:00 2001
From: Theo <theo@iconara.net>
Date: Sat, 8 Nov 2014 18:47:27 +0100
Subject: [PATCH 16/23] Rename #each and @listeners #subscribe and @subscribers

---
 lib/ione/io/base_connection.rb |  2 +-
 lib/ione/stream.rb             | 22 +++++++++++-----------
 spec/ione/stream_spec.rb       | 10 +++++-----
 3 files changed, 17 insertions(+), 17 deletions(-)

diff --git a/lib/ione/io/base_connection.rb b/lib/ione/io/base_connection.rb
index 500932b..85eb149 100644
--- a/lib/ione/io/base_connection.rb
+++ b/lib/ione/io/base_connection.rb
@@ -127,7 +127,7 @@ def to_stream
       #
       # @yield [String] the new data
       def on_data(&listener)
-        @data_stream.each(&listener)
+        @data_stream.subscribe(listener)
         nil
       end
 
diff --git a/lib/ione/stream.rb b/lib/ione/stream.rb
index 2de4c80..6458eab 100644
--- a/lib/ione/stream.rb
+++ b/lib/ione/stream.rb
@@ -6,28 +6,28 @@ module Ione
   class Stream
     # @private
     def initialize
-      @listeners = []
+      @subscribers = []
       @lock = Mutex.new
     end
 
     # @yieldparam [Object] element each element that flows through the stream
     # @return [self] the stream itself
-    def subscribe(listener=nil, &block)
+    def subscribe(subscriber=nil, &block)
       @lock.lock
-      listeners = @listeners.dup
-      listeners << (listener || block)
-      @listeners = listeners
+      subscribers = @subscribers.dup
+      subscribers << (subscriber || block)
+      @subscribers = subscribers
       self
     ensure
       @lock.unlock
     end
     alias_method :each, :subscribe
 
-    def unsubscribe(listener)
+    def unsubscribe(subscriber)
       @lock.lock
-      listeners = @listeners.dup
-      listeners.delete(listener)
-      @listeners = listeners
+      subscribers = @subscribers.dup
+      subscribers.delete(subscriber)
+      @subscribers = subscribers
     ensure
       @lock.unlock
     end
@@ -35,8 +35,8 @@ def unsubscribe(listener)
     private
 
     def deliver(element)
-      @listeners.each do |listener|
-        listener.call(element) rescue nil
+      @subscribers.each do |subscriber|
+        subscriber.call(element) rescue nil
       end
       self
     end
diff --git a/spec/ione/stream_spec.rb b/spec/ione/stream_spec.rb
index 8c59bef..8d8ec42 100644
--- a/spec/ione/stream_spec.rb
+++ b/spec/ione/stream_spec.rb
@@ -53,22 +53,22 @@ class Stream
         end
       end
 
-      describe '#each' do
+      describe '#subscribe' do
         it 'yields each element that is pushed to the stream' do
           pushed_elements = []
-          stream.each { |e| pushed_elements << e }
+          stream.subscribe { |e| pushed_elements << e }
           stream.push('foo')
           stream.push('bar')
           pushed_elements.should == ['foo', 'bar']
         end
 
         it 'returns self' do
-          stream.each { }.should equal(stream)
+          stream.subscribe { }.should equal(stream)
         end
 
-        it 'is aliased as #subscribe' do
+        it 'is aliased as #each' do
           pushed_elements = []
-          stream.subscribe { |e| pushed_elements << e }
+          stream.each { |e| pushed_elements << e }
           stream.push('foo')
           stream.push('bar')
           pushed_elements.should == ['foo', 'bar']

From b9cf0a4e3580c3bc14ff8087bcfed88268012fff Mon Sep 17 00:00:00 2001
From: Theo <theo@iconara.net>
Date: Sat, 8 Nov 2014 18:55:54 +0100
Subject: [PATCH 17/23] Reimplement Acceptor#on_accept as a stream

---
 lib/ione/io/acceptor.rb       | 17 +++++++----------
 lib/ione/io/ssl_acceptor.rb   |  2 +-
 spec/ione/io/acceptor_spec.rb | 17 +++++++++++++++++
 3 files changed, 25 insertions(+), 11 deletions(-)

diff --git a/lib/ione/io/acceptor.rb b/lib/ione/io/acceptor.rb
index 6b78767..cf088ea 100644
--- a/lib/ione/io/acceptor.rb
+++ b/lib/ione/io/acceptor.rb
@@ -23,18 +23,20 @@ def initialize(host, port, backlog, unblocker, reactor, socket_impl=nil)
         @unblocker = unblocker
         @reactor = reactor
         @socket_impl = socket_impl || ServerSocket
-        @accept_listeners = []
+        @accept_stream = Stream::PushStream.new
         @lock = Mutex.new
         @state = BINDING_STATE
       end
 
+      def to_stream
+        @accept_stream
+      end
+
       # Register a listener to be notified when client connections are accepted
       #
       # @yieldparam [Ione::Io::ServerConnection] the connection to the client
       def on_accept(&listener)
-        @lock.synchronize do
-          @accept_listeners << listener
-        end
+        @accept_stream.subscribe(listener)
       end
 
       # @private
@@ -106,7 +108,7 @@ def read
         client_socket, host, port = accept
         connection = ServerConnection.new(client_socket, host, port, @unblocker)
         @reactor.accept(connection)
-        notify_accept_listeners(connection)
+        @accept_stream << connection
       end
 
       if RUBY_ENGINE == 'jruby'
@@ -129,11 +131,6 @@ def accept
         port, host = @socket_impl.unpack_sockaddr_in(client_sockaddr)
         return client_socket, host, port
       end
-
-      def notify_accept_listeners(connection)
-        listeners = @lock.synchronize { @accept_listeners }
-        listeners.each { |l| l.call(connection) rescue nil }
-      end
     end
   end
 end
diff --git a/lib/ione/io/ssl_acceptor.rb b/lib/ione/io/ssl_acceptor.rb
index 6fac235..8f223b3 100644
--- a/lib/ione/io/ssl_acceptor.rb
+++ b/lib/ione/io/ssl_acceptor.rb
@@ -13,7 +13,7 @@ def initialize(host, port, backlog, unblocker, reactor, ssl_context, socket_impl
 
       def read
         client_socket, host, port = accept
-        connection = SslServerConnection.new(client_socket, host, port, @unblocker, @ssl_context, method(:notify_accept_listeners), @ssl_socket_impl)
+        connection = SslServerConnection.new(client_socket, host, port, @unblocker, @ssl_context, @accept_stream.to_proc, @ssl_socket_impl)
         @reactor.accept(connection)
       end
     end
diff --git a/spec/ione/io/acceptor_spec.rb b/spec/ione/io/acceptor_spec.rb
index 43a53b9..00005ab 100644
--- a/spec/ione/io/acceptor_spec.rb
+++ b/spec/ione/io/acceptor_spec.rb
@@ -187,6 +187,23 @@ module Io
         end
       end
 
+      describe '#to_stream' do
+        include_context 'accepting_connections'
+
+        it 'returns a a stream of accepted connections' do
+          received_connection1 = nil
+          received_connection2 = nil
+          stream = acceptor.to_stream
+          stream.subscribe { |c| received_connection1 = c }
+          stream.subscribe { |c| received_connection2 = c }
+          acceptor.bind
+          acceptor.read
+          received_connection1.host.should == 'example.com'
+          received_connection2.host.should == 'example.com'
+          received_connection2.port.should == 3333
+        end
+      end
+
       describe '#on_accept' do
         include_context 'accepting_connections'
 

From 74b76663488e741af0dd8be665a7f2f5a32df26b Mon Sep 17 00:00:00 2001
From: Theo <theo@iconara.net>
Date: Sat, 8 Nov 2014 21:42:58 +0100
Subject: [PATCH 18/23] Rename {Stream,}Combinators

Because Future::Combinators and not Future::FutureCombinators (any more)
---
 lib/ione/stream.rb | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/lib/ione/stream.rb b/lib/ione/stream.rb
index 6458eab..601cf7c 100644
--- a/lib/ione/stream.rb
+++ b/lib/ione/stream.rb
@@ -41,7 +41,7 @@ def deliver(element)
       self
     end
 
-    module StreamCombinators
+    module Combinators
       # @yieldparam [Object] element
       # @yieldreturn [Object] the transformed element
       # @return [Ione::Stream]
@@ -56,6 +56,7 @@ def select(&filter)
         FilteredStream.new(self, filter)
       end
 
+      # @param [Object] state
       # @yieldparam [Object] element
       # @yieldparam [Ione::Stream::PushStream] downstream
       # @yieldparam [Object] state
@@ -82,7 +83,7 @@ def drop(n)
       end
     end
 
-    include StreamCombinators
+    include Combinators
 
     class PushStream < Stream
       module_eval do

From a0bd9e16989aca877114a60af4945530fbb23427 Mon Sep 17 00:00:00 2001
From: Theo <theo@iconara.net>
Date: Sun, 9 Nov 2014 14:02:33 +0100
Subject: [PATCH 19/23] Change the description of some of the Stream tests
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

I’m not really sure how to formulate it, but this feels better than the current wording.
---
 spec/ione/stream_spec.rb | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/spec/ione/stream_spec.rb b/spec/ione/stream_spec.rb
index 8d8ec42..9985d7e 100644
--- a/spec/ione/stream_spec.rb
+++ b/spec/ione/stream_spec.rb
@@ -88,7 +88,7 @@ class Stream
       end
 
       describe '#map' do
-        it 'returns a stream that will yield elements transformed by the specified block' do
+        it 'returns a stream of elements transformed by the specified block' do
           pushed_elements = []
           transformed_stream = stream.map { |e| e.reverse }
           transformed_stream.each { |e| pushed_elements << e }
@@ -99,7 +99,7 @@ class Stream
       end
 
       describe '#select' do
-        it 'returns a stream that will yield only the elements for which the specified block returns true' do
+        it 'returns a stream of the elements for which the specified block returns true' do
           pushed_elements = []
           filtered_stream = stream.select { |e| e.include?('a') }
           filtered_stream.each { |e| pushed_elements << e }
@@ -112,7 +112,7 @@ class Stream
       end
 
       describe '#aggregate' do
-        it 'returns a stream that will yield new elements produced by the specified block' do
+        it 'returns a stream of new elements produced by the specified block' do
           pushed_elements = []
           sum = 0
           aggregate_stream = stream.aggregate do |e, downstream|
@@ -142,7 +142,7 @@ class Stream
       end
 
       describe '#take' do
-        it 'returns a stream that will yield only the specified number of elements' do
+        it 'returns a stream of only the specified number of elements' do
           pushed_elements = []
           filtered_stream = stream.take(3)
           filtered_stream.each { |e| pushed_elements << e }

From 8395bba83293b7a2ef175e313ae0f27f4df66edc Mon Sep 17 00:00:00 2001
From: Theo <theo@iconara.net>
Date: Sun, 9 Nov 2014 14:03:25 +0100
Subject: [PATCH 20/23] Remove bad API docs from Stream#take and #drop

---
 lib/ione/stream.rb | 2 --
 1 file changed, 2 deletions(-)

diff --git a/lib/ione/stream.rb b/lib/ione/stream.rb
index 601cf7c..17c84ad 100644
--- a/lib/ione/stream.rb
+++ b/lib/ione/stream.rb
@@ -68,7 +68,6 @@ def aggregate(state=nil, &aggregator)
 
       # @param [Integer] n the number of elements to pass downstream before
       #   unsubscribing
-      # @yieldparam [Object] element
       # @return [Ione::Stream]
       def take(n)
         LimitedStream.new(self, n)
@@ -76,7 +75,6 @@ def take(n)
 
       # @param [Integer] n the number of elements to skip before passing
       #   elements downstream
-      # @yieldparam [Object] element
       # @return [Ione::Stream]
       def drop(n)
         SkippingStream.new(self, n)

From 25c46ad46c8fd5c724956e163f161e4d98e58762 Mon Sep 17 00:00:00 2001
From: Theo <theo@iconara.net>
Date: Sun, 9 Nov 2014 14:14:50 +0100
Subject: [PATCH 21/23] Make some small changes to the HTTP client example

---
 examples/http_client/lib/ione/http_client.rb | 33 ++++++++++++--------
 1 file changed, 20 insertions(+), 13 deletions(-)

diff --git a/examples/http_client/lib/ione/http_client.rb b/examples/http_client/lib/ione/http_client.rb
index 75b21b3..b33ead9 100644
--- a/examples/http_client/lib/ione/http_client.rb
+++ b/examples/http_client/lib/ione/http_client.rb
@@ -34,7 +34,9 @@ def get(url, headers={})
         ctx.cert_store = @cert_store
         options[:ssl] = ctx
       end
-      f = @reactor.connect(uri.host, uri.port, options) { |connection| HttpProtocolHandler.new(connection) }
+      f = @reactor.connect(uri.host, uri.port, options) do |connection|
+        HttpProtocolHandler.new(connection)
+      end
       f.flat_map do |handler|
         handler.send_get(uri.path, uri.query, headers)
       end
@@ -50,18 +52,22 @@ def initialize(connection)
     end
 
     def send_get(path, query, headers)
-      message = 'GET '
-      message << path
-      message << '?' << query if query && !query.empty?
-      message << " HTTP/1.1\r\n"
-      headers.each do |key, value|
-        message << key
-        message << ': '
-        message << value
-        message << "\r\n"
+      @connection.write do |buffer|
+        buffer << 'GET '
+        buffer << path
+        if query && !query.empty?
+          buffer << '?'
+          buffer << query
+        end
+        buffer << " HTTP/1.1\r\n"
+        headers.each do |key, value|
+          buffer << key
+          buffer << ':'
+          buffer << value
+          buffer << "\r\n"
+        end
+        buffer << "\r\n"
       end
-      message << "\r\n"
-      @connection.write(message)
       @promises << Promise.new
       @promises.last.future
     end
@@ -84,7 +90,8 @@ def on_body(chunk)
     end
 
     def on_message_complete
-      @promises.shift.fulfill(HttpResponse.new(@http_parser.status_code, @headers, @body))
+      response = HttpResponse.new(@http_parser.status_code, @headers, @body)
+      @promises.shift.fulfill(response)
     end
   end
 

From 9bcb7394549091eb5620109d8fe287e55ad0c457 Mon Sep 17 00:00:00 2001
From: Theo <theo@iconara.net>
Date: Sun, 9 Nov 2014 14:14:57 +0100
Subject: [PATCH 22/23] Use streams in the HTTP client example

---
 examples/http_client/lib/ione/http_client.rb | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/examples/http_client/lib/ione/http_client.rb b/examples/http_client/lib/ione/http_client.rb
index b33ead9..79bf679 100644
--- a/examples/http_client/lib/ione/http_client.rb
+++ b/examples/http_client/lib/ione/http_client.rb
@@ -46,7 +46,7 @@ def get(url, headers={})
   class HttpProtocolHandler
     def initialize(connection)
       @connection = connection
-      @connection.on_data(&method(:process_data))
+      @connection.to_stream.subscribe(method(:process_data))
       @http_parser = Http::Parser.new(self)
       @promises = []
     end

From 9f76e1f2fe6630449bf622c84e69073d0abe0528 Mon Sep 17 00:00:00 2001
From: Theo <theo@iconara.net>
Date: Tue, 30 Dec 2014 15:59:38 +0100
Subject: [PATCH 23/23] Change things around to get closer to the reactive
 streams spec

* Stream#subscribe now takes a Subscriber
* PushStream is renamed Source and has a #<< for publishing
* #subscribe and #unsubscribe are encapsulated in Publisher
* The private #deliver method of Stream has been encapsulated in a class called Processor, which is both a Publisher and a Subscriber, and rename it #call
* The combinators have been rewritten to be Sources
---
 lib/ione/io/acceptor.rb        |   2 +-
 lib/ione/io/base_connection.rb |   2 +-
 lib/ione/stream.rb             | 182 ++++++++++++++--------------
 spec/ione/stream_spec.rb       | 209 +++++++++++++++++++++------------
 4 files changed, 236 insertions(+), 159 deletions(-)

diff --git a/lib/ione/io/acceptor.rb b/lib/ione/io/acceptor.rb
index cf088ea..d8fe56b 100644
--- a/lib/ione/io/acceptor.rb
+++ b/lib/ione/io/acceptor.rb
@@ -23,7 +23,7 @@ def initialize(host, port, backlog, unblocker, reactor, socket_impl=nil)
         @unblocker = unblocker
         @reactor = reactor
         @socket_impl = socket_impl || ServerSocket
-        @accept_stream = Stream::PushStream.new
+        @accept_stream = Stream::Source.new
         @lock = Mutex.new
         @state = BINDING_STATE
       end
diff --git a/lib/ione/io/base_connection.rb b/lib/ione/io/base_connection.rb
index 85eb149..eaa0d81 100644
--- a/lib/ione/io/base_connection.rb
+++ b/lib/ione/io/base_connection.rb
@@ -21,7 +21,7 @@ def initialize(host, port, unblocker)
         @lock = Mutex.new
         @write_buffer = ByteBuffer.new
         @closed_promise = Promise.new
-        @data_stream = Stream::PushStream.new
+        @data_stream = Stream::Source.new
       end
 
       # Closes the connection
diff --git a/lib/ione/stream.rb b/lib/ione/stream.rb
index 17c84ad..e6b6619 100644
--- a/lib/ione/stream.rb
+++ b/lib/ione/stream.rb
@@ -2,43 +2,47 @@
 
 module Ione
   # @abstract Base class for streams
-  # @see Ione::Stream::PushStream
+  # @see Ione::Stream::Publisher
   class Stream
-    # @private
-    def initialize
-      @subscribers = []
-      @lock = Mutex.new
-    end
+    class Publisher
+      # @private
+      def initialize
+        @subscribers = {}
+        @lock = Mutex.new
+      end
 
-    # @yieldparam [Object] element each element that flows through the stream
-    # @return [self] the stream itself
-    def subscribe(subscriber=nil, &block)
-      @lock.lock
-      subscribers = @subscribers.dup
-      subscribers << (subscriber || block)
-      @subscribers = subscribers
-      self
-    ensure
-      @lock.unlock
-    end
-    alias_method :each, :subscribe
-
-    def unsubscribe(subscriber)
-      @lock.lock
-      subscribers = @subscribers.dup
-      subscribers.delete(subscriber)
-      @subscribers = subscribers
-    ensure
-      @lock.unlock
+      # @yieldparam [Object] element each element that flows through the stream
+      # @return [self] the stream itself
+      def subscribe(subscriber=nil, &block)
+        subscriber ||= block
+        unless subscriber.respond_to?(:call)
+          raise ArgumentError, %(A subscriber must respond to #call)
+        end
+        @lock.lock
+        begin
+          @subscribers[subscriber] = 1
+          self
+        ensure
+          @lock.unlock
+        end
+        subscriber
+      end
+      alias_method :each, :subscribe
+
+      # @param [Object] subscriber
+      # @return [self] the stream itself
+      def unsubscribe(subscriber)
+        @lock.lock
+        @subscribers.delete(subscriber)
+        subscriber
+      ensure
+        @lock.unlock
+      end
     end
 
-    private
-
-    def deliver(element)
-      @subscribers.each do |subscriber|
-        subscriber.call(element) rescue nil
+    module Subscriber
+      def call(element)
       end
-      self
     end
 
     module Combinators
@@ -46,116 +50,124 @@ module Combinators
       # @yieldreturn [Object] the transformed element
       # @return [Ione::Stream]
       def map(&transformer)
-        TransformedStream.new(self, transformer)
+        subscribe(TransformedStream.new(transformer))
       end
 
       # @yieldparam [Object] element
       # @yieldreturn [Boolean] whether or not to pass the element downstream
       # @return [Ione::Stream]
       def select(&filter)
-        FilteredStream.new(self, filter)
+        subscribe(FilteredStream.new(filter))
       end
 
       # @param [Object] state
       # @yieldparam [Object] element
-      # @yieldparam [Ione::Stream::PushStream] downstream
+      # @yieldparam [Ione::Stream::Publisher] downstream
       # @yieldparam [Object] state
       # @yieldreturn [Object] the next state
       # @return [Ione::Stream]
       def aggregate(state=nil, &aggregator)
-        AggregatingStream.new(self, aggregator, state)
+        subscribe(AggregatingStream.new(aggregator, state))
       end
 
       # @param [Integer] n the number of elements to pass downstream before
       #   unsubscribing
       # @return [Ione::Stream]
       def take(n)
-        LimitedStream.new(self, n)
+        subscribe(LimitedStream.new(self, n))
       end
 
       # @param [Integer] n the number of elements to skip before passing
       #   elements downstream
       # @return [Ione::Stream]
       def drop(n)
-        SkippingStream.new(self, n)
+        subscribe(SkippingStream.new(n))
       end
     end
 
-    include Combinators
-
-    class PushStream < Stream
-      module_eval do
-        # this crazyness is just to hide these declarations from Yard
-        alias_method :push, :deliver
-        public :push
-        alias_method :<<, :deliver
-        public :<<
-      end
-
-      # @!parse
-      #   # @param [Object] element
-      #   # @return [self]
-      #   def push(element); end
-      #   alias_method :<<, :push
+    class Processor < Publisher
+      include Subscriber
+      include Combinators
+    end
 
-      # @return [Proc] a Proc that can be used to push elements to this stream
-      def to_proc
-        method(:push).to_proc
+    class Source < Processor
+      def <<(element)
+        @subscribers.each_key do |subscriber|
+          subscriber.call(element) rescue nil
+        end
+        element
       end
     end
 
     # @private
-    class TransformedStream < Stream
-      def initialize(upstream, transformer)
+    class TransformedStream < Source
+      def initialize(transformer)
         super()
-        upstream.each { |e| deliver(transformer.call(e)) }
+        @transformer = transformer
+      end
+
+      def call(element)
+        self << @transformer.call(element)
       end
     end
 
     # @private
-    class FilteredStream < Stream
-      def initialize(upstream, filter)
+    class FilteredStream < Source
+      def initialize(filter)
         super()
-        upstream.each { |e| deliver(e) if filter.call(e) }
+        @filter = filter
+      end
+
+      def call(element)
+        self << element if @filter.call(element)
       end
     end
 
     # @private
-    class AggregatingStream < PushStream
-      def initialize(upstream, aggregator, state)
+    class AggregatingStream < Source
+      def initialize(aggregator, state)
         super()
-        upstream.each { |e| state = aggregator.call(e, self, state) }
+        @aggregator = aggregator
+        @state = state
+      end
+
+      def call(element)
+        @state = @aggregator.call(element, self, @state)
       end
     end
 
     # @private
-    class LimitedStream < Stream
+    class LimitedStream < Source
       def initialize(upstream, n)
         super()
-        counter = 0
-        subscriber = proc do |e|
-          if counter < n
-            deliver(e)
-          else
-            upstream.unsubscribe(subscriber)
-          end
-          counter += 1
+        @upstream = upstream
+        @counter = 0
+        @limit = n
+      end
+
+      def call(element)
+        if @counter < @limit
+          self << element
+        else
+          @upstream.unsubscribe(self)
         end
-        upstream.subscribe(subscriber)
+        @counter += 1
       end
     end
 
     # @private
-    class SkippingStream < Stream
-      def initialize(upstream, n)
+    class SkippingStream < Source
+      def initialize(n)
         super()
-        counter = 0
-        upstream.subscribe do |e|
-          if counter == n
-            deliver(e)
-          else
-            counter += 1
-          end
+        @counter = 0
+        @skips = n
+      end
+
+      def call(element)
+        if @counter == @skips
+          self << element
+        else
+          @counter += 1
         end
       end
     end
diff --git a/spec/ione/stream_spec.rb b/spec/ione/stream_spec.rb
index 9985d7e..cc89718 100644
--- a/spec/ione/stream_spec.rb
+++ b/spec/ione/stream_spec.rb
@@ -5,127 +5,176 @@
 
 module Ione
   class Stream
-    describe PushStream do
+    describe Source do
       let :stream do
         described_class.new
       end
 
-      describe '#push' do
-        it 'pushes an element to the stream' do
-          pushed_elements = []
-          stream.each { |e| pushed_elements << e }
-          stream.push('foo')
-          pushed_elements.should == ['foo']
-        end
-
-        it 'is aliased as #<<' do
-          pushed_elements = []
-          stream.each { |e| pushed_elements << e }
+      describe '#<<' do
+        it 'publishes an element to the stream' do
+          published_elements = []
+          stream.each { |e| published_elements << e }
           stream << 'foo'
-          pushed_elements.should == ['foo']
+          published_elements.should == ['foo']
         end
 
-        it 'returns self' do
-          stream.push('foo').should equal(stream)
+        it 'returns the message' do
+          (stream << 'foo').should == 'foo'
         end
 
         it 'delivers the element to all listeners' do
-          pushed_elements1 = []
-          pushed_elements2 = []
-          pushed_elements3 = []
-          stream.each { |e| pushed_elements1 << e }
-          stream.each { |e| pushed_elements2 << e }
-          stream.each { |e| pushed_elements3 << e }
-          stream.push('foo')
-          pushed_elements1.should == ['foo']
-          pushed_elements2.should == ['foo']
-          pushed_elements3.should == ['foo']
+          published_elements1 = []
+          published_elements2 = []
+          published_elements3 = []
+          stream.each { |e| published_elements1 << e }
+          stream.each { |e| published_elements2 << e }
+          stream.each { |e| published_elements3 << e }
+          stream << 'foo'
+          published_elements1.should == ['foo']
+          published_elements2.should == ['foo']
+          published_elements3.should == ['foo']
         end
 
         it 'ignores errors raised by listeners' do
-          pushed_elements1 = []
-          pushed_elements2 = []
+          published_elements1 = []
+          published_elements2 = []
           stream.each { |e| raise 'bork!' }
-          stream.each { |e| pushed_elements2 << e }
-          stream.push('foo')
-          pushed_elements1.should == []
-          pushed_elements2.should == ['foo']
+          stream.each { |e| published_elements2 << e }
+          stream << 'foo'
+          published_elements1.should == []
+          published_elements2.should == ['foo']
         end
       end
 
       describe '#subscribe' do
-        it 'yields each element that is pushed to the stream' do
-          pushed_elements = []
-          stream.subscribe { |e| pushed_elements << e }
-          stream.push('foo')
-          stream.push('bar')
-          pushed_elements.should == ['foo', 'bar']
+        context 'when given a block' do
+          it 'yields each element that is pushed to the stream' do
+            yielded_elements = []
+            stream.subscribe { |e| yielded_elements << e }
+            stream << 'foo'
+            stream << 'bar'
+            yielded_elements.should == ['foo', 'bar']
+          end
+
+          it 'returns the subscriber' do
+            stream.subscribe { }.should be_a(Proc)
+          end
         end
 
-        it 'returns self' do
-          stream.subscribe { }.should equal(stream)
+        context 'when given a subscriber' do
+          let :subscriber do
+            StreamSpec::Subscriber.new
+          end
+
+          it 'delivers each element that is pushed to the stream' do
+            stream.subscribe(subscriber)
+            stream << 'foo'
+            stream << 'bar'
+            subscriber.received_elements.should == ['foo', 'bar']
+          end
+
+          it 'returns the subscriber' do
+            stream.subscribe(subscriber).should equal(subscriber)
+          end
         end
 
-        it 'is aliased as #each' do
-          pushed_elements = []
-          stream.each { |e| pushed_elements << e }
-          stream.push('foo')
-          stream.push('bar')
-          pushed_elements.should == ['foo', 'bar']
+        context 'when given a callable' do
+          it 'delivers each element that is pushed to the stream' do
+            delivered_elements = []
+            stream.subscribe(proc { |e| delivered_elements << e })
+            stream << 'foo'
+            stream << 'bar'
+            delivered_elements.should == ['foo', 'bar']
+          end
+
+          it 'returns the subscriber' do
+            subscriber = proc { |e| delivered_elements << e }
+            stream.subscribe(subscriber).should equal(subscriber)
+          end
         end
-      end
 
-      describe '#to_proc' do
-        it 'returns a Proc that can be used to push elements to the stream' do
-          pushed_elements = []
-          another_stream = described_class.new
-          another_stream.each { |e| pushed_elements << e}
-          stream.each(&another_stream)
+        context 'when given something that does not respond to neither #receive nor #call' do
+          it 'raises ArgumentError' do
+            expect { stream.subscribe('foo') }.to raise_error(ArgumentError)
+            expect { stream.subscribe }.to raise_error(ArgumentError)
+          end
+        end
+
+        it 'is aliased as #each' do
+          published_elements = []
+          stream.each { |e| published_elements << e }
           stream << 'foo'
           stream << 'bar'
-          pushed_elements.should == ['foo', 'bar']
+          published_elements.should == ['foo', 'bar']
+        end
+      end
+
+      describe '#unsubscribe' do
+        context 'with a Proc subscriber' do
+          it 'stops delivering messages to the subscriber' do
+            delivered_elements = []
+            subscriber = proc { |e| delivered_elements << e }
+            stream.subscribe(subscriber)
+            stream << 'foo'
+            stream << 'bar'
+            stream.unsubscribe(subscriber)
+            stream << 'baz'
+            delivered_elements.should_not include('baz')
+          end
+        end
+
+        context 'with a Subscriber subscriber' do
+          it 'stops delivering messages to the subscriber' do
+            subscriber = StreamSpec::Subscriber.new
+            stream.subscribe(subscriber)
+            stream << 'foo'
+            stream << 'bar'
+            stream.unsubscribe(subscriber)
+            stream << 'baz'
+            subscriber.received_elements.should_not include('baz')
+          end
         end
       end
 
       describe '#map' do
         it 'returns a stream of elements transformed by the specified block' do
-          pushed_elements = []
+          published_elements = []
           transformed_stream = stream.map { |e| e.reverse }
-          transformed_stream.each { |e| pushed_elements << e }
+          transformed_stream.each { |e| published_elements << e }
           stream << 'foo'
           stream << 'bar'
-          pushed_elements.should == ['oof', 'rab']
+          published_elements.should == ['oof', 'rab']
         end
       end
 
       describe '#select' do
         it 'returns a stream of the elements for which the specified block returns true' do
-          pushed_elements = []
+          published_elements = []
           filtered_stream = stream.select { |e| e.include?('a') }
-          filtered_stream.each { |e| pushed_elements << e }
+          filtered_stream.each { |e| published_elements << e }
           stream << 'foo'
           stream << 'bar'
           stream << 'baz'
           stream << 'qux'
-          pushed_elements.should == ['bar', 'baz']
+          published_elements.should == ['bar', 'baz']
         end
       end
 
       describe '#aggregate' do
         it 'returns a stream of new elements produced by the specified block' do
-          pushed_elements = []
+          published_elements = []
           sum = 0
           aggregate_stream = stream.aggregate do |e, downstream|
             sum += e
             downstream << sum
           end
-          aggregate_stream.each { |e| pushed_elements << e }
-          1.upto(5) { |n| stream.push(n) }
-          pushed_elements.should == [1, 1 + 2, 1 + 2 + 3, 1 + 2 + 3 + 4, 1 + 2 + 3 + 4 + 5]
+          aggregate_stream.each { |e| published_elements << e }
+          1.upto(5) { |n| stream << n }
+          published_elements.should == [1, 1 + 2, 1 + 2 + 3, 1 + 2 + 3 + 4, 1 + 2 + 3 + 4 + 5]
         end
 
         it 'passes the given argument to the first invocation of the block, and the block\'s return value on each subsequent invocation' do
-          pushed_elements = []
+          published_elements = []
           aggregate_stream = stream.aggregate('') do |e, downstream, buffer|
             buffer << e
             while (i = buffer.index("\n"))
@@ -133,39 +182,55 @@ class Stream
             end
             buffer
           end
-          aggregate_stream.each { |e| pushed_elements << e }
+          aggregate_stream.each { |e| published_elements << e }
           stream << "fo"
           stream << "o\nbar\nba"
           stream << "z\n"
-          pushed_elements.should == ["foo\n", "bar\n", "baz\n"]
+          published_elements.should == ["foo\n", "bar\n", "baz\n"]
         end
       end
 
       describe '#take' do
         it 'returns a stream of only the specified number of elements' do
-          pushed_elements = []
+          published_elements = []
           filtered_stream = stream.take(3)
-          filtered_stream.each { |e| pushed_elements << e }
+          filtered_stream.each { |e| published_elements << e }
           stream << 'foo'
           stream << 'bar'
           stream << 'baz'
           stream << 'qux'
-          pushed_elements.should == ['foo', 'bar', 'baz']
+          published_elements.should == ['foo', 'bar', 'baz']
         end
       end
 
       describe '#drop' do
         it 'returns a stream that will skip the specified number of items' do
-          pushed_elements = []
+          published_elements = []
           filtered_stream = stream.drop(2)
-          filtered_stream.each { |e| pushed_elements << e }
+          filtered_stream.each { |e| published_elements << e }
           stream << 'foo'
           stream << 'bar'
           stream << 'baz'
           stream << 'qux'
-          pushed_elements.should == ['baz', 'qux']
+          published_elements.should == ['baz', 'qux']
         end
       end
     end
   end
 end
+
+module StreamSpec
+  class Subscriber
+    include Ione::Stream::Subscriber
+
+    attr_reader :received_elements
+
+    def initialize
+      @received_elements = []
+    end
+
+    def call(element)
+      @received_elements << element
+    end
+  end
+end
\ No newline at end of file