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.
diff --git a/examples/http_client/lib/ione/http_client.rb b/examples/http_client/lib/ione/http_client.rb
index 75b21b3..79bf679 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
@@ -44,24 +46,28 @@ 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
 
     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
 
diff --git a/examples/redis_client/lib/ione/redis_client.rb b/examples/redis_client/lib/ione/redis_client.rb
index f931d46..f4aeffd 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
@@ -13,60 +15,55 @@ 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_data_chunks(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)
+    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)
+          line.chomp!
+          downstream << line
+        end
+        buffer
+      end
+      process_lines(line_stream)
     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
-      end
-      lines.each do |line|
-        @line_listener.call(line) if @line_listener
+    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?
+          downstream << [state.response, state.error?]
+        end
+        state
       end
+      process_responses(response_stream)
     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))
+    def process_responses(response_stream)
+      response_stream.each do |response, error|
+        promise = @responses.shift
+        if error
+          promise.fail(RedisError.new(response))
+        else
+          promise.fulfill(response)
+        end
+      end
     end
 
-    def send_request(*args)
+    def method_missing(*args)
       promise = Ione::Promise.new
       @responses << promise
       request = "*#{args.size}\r\n"
@@ -74,87 +71,107 @@ 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
-
-    def handle_line(line)
-      @state = @state.handle_line(line)
-    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
+        def response?
+          false
+        end
 
-    class State
-      def initialize(result_handler)
-        @result_handler = result_handler
-      end
+        private
 
-      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(size)
+          @size = size
+          @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 == @size
+            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
diff --git a/examples/redis_client/spec/ione/redis_client_spec.rb b/examples/redis_client/spec/ione/redis_client_spec.rb
index 39a167e..4c25905 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,16 @@ module Ione
     end
 
     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 '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
-      pending('Redis not running', unless: client)
       f = client.del('stuff')
       f.value
       f = client.rpush('stuff', 'hello', 'world')
@@ -60,7 +63,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')
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/io/acceptor.rb b/lib/ione/io/acceptor.rb
index 6b78767..d8fe56b 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::Source.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/base_connection.rb b/lib/ione/io/base_connection.rb
index 2904011..eaa0d81 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::Source.new
       end
 
       # Closes the connection
@@ -84,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.
@@ -103,7 +127,8 @@ def writable?
       #
       # @yield [String] the new data
       def on_data(&listener)
-        @data_listener = listener
+        @data_stream.subscribe(listener)
+        nil
       end
 
       # Register to receive a notification when the socket is closed, both for
@@ -166,8 +191,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_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/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
diff --git a/lib/ione/stream.rb b/lib/ione/stream.rb
new file mode 100644
index 0000000..e6b6619
--- /dev/null
+++ b/lib/ione/stream.rb
@@ -0,0 +1,175 @@
+# encoding: utf-8
+
+module Ione
+  # @abstract Base class for streams
+  # @see Ione::Stream::Publisher
+  class Stream
+    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)
+        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
+
+    module Subscriber
+      def call(element)
+      end
+    end
+
+    module Combinators
+      # @yieldparam [Object] element
+      # @yieldreturn [Object] the transformed element
+      # @return [Ione::Stream]
+      def map(&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)
+        subscribe(FilteredStream.new(filter))
+      end
+
+      # @param [Object] state
+      # @yieldparam [Object] element
+      # @yieldparam [Ione::Stream::Publisher] downstream
+      # @yieldparam [Object] state
+      # @yieldreturn [Object] the next state
+      # @return [Ione::Stream]
+      def aggregate(state=nil, &aggregator)
+        subscribe(AggregatingStream.new(aggregator, state))
+      end
+
+      # @param [Integer] n the number of elements to pass downstream before
+      #   unsubscribing
+      # @return [Ione::Stream]
+      def take(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)
+        subscribe(SkippingStream.new(n))
+      end
+    end
+
+    class Processor < Publisher
+      include Subscriber
+      include Combinators
+    end
+
+    class Source < Processor
+      def <<(element)
+        @subscribers.each_key do |subscriber|
+          subscriber.call(element) rescue nil
+        end
+        element
+      end
+    end
+
+    # @private
+    class TransformedStream < Source
+      def initialize(transformer)
+        super()
+        @transformer = transformer
+      end
+
+      def call(element)
+        self << @transformer.call(element)
+      end
+    end
+
+    # @private
+    class FilteredStream < Source
+      def initialize(filter)
+        super()
+        @filter = filter
+      end
+
+      def call(element)
+        self << element if @filter.call(element)
+      end
+    end
+
+    # @private
+    class AggregatingStream < Source
+      def initialize(aggregator, state)
+        super()
+        @aggregator = aggregator
+        @state = state
+      end
+
+      def call(element)
+        @state = @aggregator.call(element, self, @state)
+      end
+    end
+
+    # @private
+    class LimitedStream < Source
+      def initialize(upstream, n)
+        super()
+        @upstream = upstream
+        @counter = 0
+        @limit = n
+      end
+
+      def call(element)
+        if @counter < @limit
+          self << element
+        else
+          @upstream.unsubscribe(self)
+        end
+        @counter += 1
+      end
+    end
+
+    # @private
+    class SkippingStream < Source
+      def initialize(n)
+        super()
+        @counter = 0
+        @skips = n
+      end
+
+      def call(element)
+        if @counter == @skips
+          self << element
+        else
+          @counter += 1
+        end
+      end
+    end
+  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'
 
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
diff --git a/spec/ione/stream_spec.rb b/spec/ione/stream_spec.rb
new file mode 100644
index 0000000..cc89718
--- /dev/null
+++ b/spec/ione/stream_spec.rb
@@ -0,0 +1,236 @@
+# encoding: utf-8
+
+require 'spec_helper'
+
+
+module Ione
+  class Stream
+    describe Source do
+      let :stream do
+        described_class.new
+      end
+
+      describe '#<<' do
+        it 'publishes an element to the stream' do
+          published_elements = []
+          stream.each { |e| published_elements << e }
+          stream << 'foo'
+          published_elements.should == ['foo']
+        end
+
+        it 'returns the message' do
+          (stream << 'foo').should == 'foo'
+        end
+
+        it 'delivers the element to all listeners' do
+          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
+          published_elements1 = []
+          published_elements2 = []
+          stream.each { |e| raise 'bork!' }
+          stream.each { |e| published_elements2 << e }
+          stream << 'foo'
+          published_elements1.should == []
+          published_elements2.should == ['foo']
+        end
+      end
+
+      describe '#subscribe' do
+        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
+
+        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
+
+        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
+
+        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'
+          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
+          published_elements = []
+          transformed_stream = stream.map { |e| e.reverse }
+          transformed_stream.each { |e| published_elements << e }
+          stream << 'foo'
+          stream << 'bar'
+          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
+          published_elements = []
+          filtered_stream = stream.select { |e| e.include?('a') }
+          filtered_stream.each { |e| published_elements << e }
+          stream << 'foo'
+          stream << 'bar'
+          stream << 'baz'
+          stream << 'qux'
+          published_elements.should == ['bar', 'baz']
+        end
+      end
+
+      describe '#aggregate' do
+        it 'returns a stream of new elements produced by the specified block' do
+          published_elements = []
+          sum = 0
+          aggregate_stream = stream.aggregate do |e, downstream|
+            sum += e
+            downstream << sum
+          end
+          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
+          published_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| published_elements << e }
+          stream << "fo"
+          stream << "o\nbar\nba"
+          stream << "z\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
+          published_elements = []
+          filtered_stream = stream.take(3)
+          filtered_stream.each { |e| published_elements << e }
+          stream << 'foo'
+          stream << 'bar'
+          stream << 'baz'
+          stream << 'qux'
+          published_elements.should == ['foo', 'bar', 'baz']
+        end
+      end
+
+      describe '#drop' do
+        it 'returns a stream that will skip the specified number of items' do
+          published_elements = []
+          filtered_stream = stream.drop(2)
+          filtered_stream.each { |e| published_elements << e }
+          stream << 'foo'
+          stream << 'bar'
+          stream << 'baz'
+          stream << '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