diff --git a/src/event_loop.rs b/src/event_loop.rs
index d04a3e4d2..3bd8ca2a0 100644
--- a/src/event_loop.rs
+++ b/src/event_loop.rs
@@ -80,7 +80,7 @@ pub struct LoopPin {
 }
 
 struct Scheduled {
-    source: IoSource,
+    readiness: Arc<AtomicUsize>,
     reader: Option<TaskHandle>,
     writer: Option<TaskHandle>,
 }
@@ -97,7 +97,6 @@ enum Direction {
 }
 
 enum Message {
-    AddSource(IoSource, Arc<Slot<io::Result<usize>>>),
     DropSource(usize),
     Schedule(usize, TaskHandle, Direction),
     AddTimeout(Instant, Arc<Slot<io::Result<TimeoutToken>>>),
@@ -107,29 +106,6 @@ enum Message {
     Drop(DropBox<dropbox::MyDrop>),
 }
 
-/// Type of I/O objects inserted into the event loop, created by `Source::new`.
-pub struct Source<E: ?Sized> {
-    readiness: AtomicUsize,
-    io: E,
-}
-
-/// I/O objects inserted into the event loop
-pub type IoSource = Arc<Source<mio::Evented + Sync + Send>>;
-
-fn register(poll: &mio::Poll,
-            token: usize,
-            sched: &Scheduled) -> io::Result<()> {
-    poll.register(&sched.source.io,
-                  mio::Token(token),
-                  mio::EventSet::readable() | mio::EventSet::writable(),
-                  mio::PollOpt::edge())
-}
-
-fn deregister(poll: &mio::Poll, sched: &Scheduled) {
-    // TODO: handle error
-    poll.deregister(&sched.source.io).unwrap();
-}
-
 impl Loop {
     /// Creates a new event loop, returning any error that happened during the
     /// creation.
@@ -334,11 +310,11 @@ impl Loop {
                 if let Some(sched) = self.dispatch.borrow_mut().get_mut(token) {
                     if event.kind().is_readable() {
                         reader = sched.reader.take();
-                        sched.source.readiness.fetch_or(1, Ordering::Relaxed);
+                        sched.readiness.fetch_or(1, Ordering::Relaxed);
                     }
                     if event.kind().is_writable() {
                         writer = sched.writer.take();
-                        sched.source.readiness.fetch_or(2, Ordering::Relaxed);
+                        sched.readiness.fetch_or(2, Ordering::Relaxed);
                     }
                 } else {
                     debug!("notified on {} which no longer exists", token);
@@ -382,10 +358,10 @@ impl Loop {
         CURRENT_LOOP.set(&self, || handle.unpark());
     }
 
-    fn add_source(&self, source: IoSource) -> io::Result<usize> {
+    fn add_source(&self, source: &mio::Evented) -> io::Result<IoToken> {
         debug!("adding a new I/O source");
         let sched = Scheduled {
-            source: source,
+            readiness: Arc::new(AtomicUsize::new(0)),
             reader: None,
             writer: None,
         };
@@ -395,14 +371,20 @@ impl Loop {
             dispatch.grow(amt);
         }
         let entry = dispatch.vacant_entry().unwrap();
-        try!(register(&self.io, entry.index(), &sched));
-        Ok(entry.insert(sched).index())
+        try!(self.io.register(source,
+                              mio::Token(entry.index()),
+                              mio::EventSet::readable() |
+                                mio::EventSet::writable(),
+                              mio::PollOpt::edge()));
+        Ok(IoToken {
+            readiness: sched.readiness.clone(),
+            token: entry.insert(sched).index(),
+        })
     }
 
     fn drop_source(&self, token: usize) {
         debug!("dropping I/O source: {}", token);
-        let sched = self.dispatch.borrow_mut().remove(token).unwrap();
-        deregister(&self.io, &sched);
+        self.dispatch.borrow_mut().remove(token).unwrap();
     }
 
     fn schedule(&self, token: usize, wake: TaskHandle, dir: Direction) {
@@ -414,10 +396,10 @@ impl Loop {
                 Direction::Read => (&mut sched.reader, 1),
                 Direction::Write => (&mut sched.writer, 2),
             };
-            let ready = sched.source.readiness.load(Ordering::SeqCst);
+            let ready = sched.readiness.load(Ordering::SeqCst);
             if ready & bit != 0 {
                 *slot = None;
-                sched.source.readiness.store(ready & !bit, Ordering::SeqCst);
+                sched.readiness.store(ready & !bit, Ordering::SeqCst);
                 Some(wake)
             } else {
                 *slot = Some(wake);
@@ -472,11 +454,6 @@ impl Loop {
 
     fn notify(&self, msg: Message) {
         match msg {
-            Message::AddSource(source, slot) => {
-                // This unwrap() should always be ok as we're the only producer
-                slot.try_produce(self.add_source(source))
-                    .ok().expect("interference with try_produce");
-            }
             Message::DropSource(tok) => self.drop_source(tok),
             Message::Schedule(tok, wake, dir) => self.schedule(tok, wake, dir),
 
@@ -545,19 +522,20 @@ impl LoopHandle {
     ///
     /// When a new I/O object is created it needs to be communicated to the
     /// event loop to ensure that it's registered and ready to receive
-    /// notifications. The event loop with then respond with a unique token that
-    /// this handle can be identified with (the resolved value of the returned
-    /// future).
+    /// notifications. The event loop with then respond back with the I/O object
+    /// and a token which can be used to send more messages to the event loop.
     ///
-    /// This token is then passed in turn to each of the methods below to
-    /// interact with notifications on the I/O object itself.
+    /// The token returned is then passed in turn to each of the methods below
+    /// to interact with notifications on the I/O object itself.
     ///
     /// # Panics
     ///
     /// The returned future will panic if the event loop this handle is
     /// associated with has gone away, or if there is an error communicating
     /// with the event loop.
-    pub fn add_source(&self, source: IoSource) -> AddSource {
+    pub fn add_source<E>(&self, source: E) -> AddSource<E>
+        where E: mio::Evented + Send + 'static,
+    {
         AddSource {
             inner: LoopFuture {
                 loop_handle: self.clone(),
@@ -567,15 +545,19 @@ impl LoopHandle {
         }
     }
 
-    /// Begin listening for read events on an event loop.
+    /// Schedule the current future task to receive a notification when the
+    /// corresponding I/O object is readable.
     ///
     /// Once an I/O object has been registered with the event loop through the
     /// `add_source` method, this method can be used with the assigned token to
-    /// begin awaiting read notifications.
+    /// notify the current future task when the next read notification comes in.
     ///
-    /// Currently the current task will be notified with *edge* semantics. This
-    /// means that whenever the underlying I/O object changes state, e.g. it was
-    /// not readable and now it is, then a notification will be sent.
+    /// The current task will only receive a notification **once** and to
+    /// receive further notifications it will need to call `schedule_read`
+    /// again.
+    ///
+    /// > **Note**: This method should generally not be used directly, but
+    /// >           rather the `ReadinessStream` type should be used instead.
     ///
     /// # Panics
     ///
@@ -585,19 +567,24 @@ impl LoopHandle {
     ///
     /// This function will also panic if there is not a currently running future
     /// task.
-    pub fn schedule_read(&self, tok: usize) {
-        self.send(Message::Schedule(tok, task::park(), Direction::Read));
+    pub fn schedule_read(&self, tok: &IoToken) {
+        self.send(Message::Schedule(tok.token, task::park(), Direction::Read));
     }
 
-    /// Begin listening for write events on an event loop.
+    /// Schedule the current future task to receive a notification when the
+    /// corresponding I/O object is writable.
     ///
     /// Once an I/O object has been registered with the event loop through the
     /// `add_source` method, this method can be used with the assigned token to
-    /// begin awaiting write notifications.
+    /// notify the current future task when the next write notification comes
+    /// in.
     ///
-    /// Currently the current task will be notified with *edge* semantics. This
-    /// means that whenever the underlying I/O object changes state, e.g. it was
-    /// not writable and now it is, then a notification will be sent.
+    /// The current task will only receive a notification **once** and to
+    /// receive further notifications it will need to call `schedule_write`
+    /// again.
+    ///
+    /// > **Note**: This method should generally not be used directly, but
+    /// >           rather the `ReadinessStream` type should be used instead.
     ///
     /// # Panics
     ///
@@ -607,8 +594,8 @@ impl LoopHandle {
     ///
     /// This function will also panic if there is not a currently running future
     /// task.
-    pub fn schedule_write(&self, tok: usize) {
-        self.send(Message::Schedule(tok, task::park(), Direction::Write));
+    pub fn schedule_write(&self, tok: &IoToken) {
+        self.send(Message::Schedule(tok.token, task::park(), Direction::Write));
     }
 
     /// Unregister all information associated with a token on an event loop,
@@ -625,13 +612,16 @@ impl LoopHandle {
     /// ensure that the callbacks are **not** invoked, so pending scheduled
     /// callbacks cannot be relied upon to get called.
     ///
+    /// > **Note**: This method should generally not be used directly, but
+    /// >           rather the `ReadinessStream` type should be used instead.
+    ///
     /// # Panics
     ///
     /// This function will panic if the event loop this handle is associated
     /// with has gone away, or if there is an error communicating with the event
     /// loop.
-    pub fn drop_source(&self, tok: usize) {
-        self.send(Message::DropSource(tok));
+    pub fn drop_source(&self, tok: &IoToken) {
+        self.send(Message::DropSource(tok.token));
     }
 
     /// Adds a new timeout to get fired at the specified instant, notifying the
@@ -731,16 +721,60 @@ impl LoopPin {
 ///
 /// Created through the `LoopHandle::add_source` method, this future can also
 /// resolve to an error if there's an issue communicating with the event loop.
-pub struct AddSource {
-    inner: LoopFuture<usize, IoSource>,
+pub struct AddSource<E> {
+    inner: LoopFuture<(E, IoToken), E>,
 }
 
-impl Future for AddSource {
-    type Item = usize;
+/// A token that identifies an active timeout.
+pub struct IoToken {
+    token: usize,
+    // TODO: can we avoid this allocation? It's kind of a bummer...
+    readiness: Arc<AtomicUsize>,
+}
+
+impl IoToken {
+	/// Consumes the last readiness notification the token this source is for
+    /// registered.
+	///
+	/// Currently sources receive readiness notifications on an edge-basis. That
+	/// is, once you receive a notification that an object can be read, you
+	/// won't receive any more notifications until all of that data has been
+	/// read.
+	///
+	/// The event loop will fill in this information and then inform futures
+	/// that they're ready to go with the `schedule` method, and then the `poll`
+	/// method can use this to figure out what happened.
+    ///
+    /// > **Note**: This method should generally not be used directly, but
+    /// >           rather the `ReadinessStream` type should be used instead.
+    // TODO: this should really return a proper newtype/enum, not a usize
+    pub fn take_readiness(&self) -> usize {
+        self.readiness.swap(0, Ordering::SeqCst)
+    }
+}
+
+impl<E> Future for AddSource<E>
+    where E: mio::Evented + Send + 'static,
+{
+    type Item = (E, IoToken);
     type Error = io::Error;
 
-    fn poll(&mut self) -> Poll<usize, io::Error> {
-        self.inner.poll(Loop::add_source, Message::AddSource)
+    fn poll(&mut self) -> Poll<(E, IoToken), io::Error> {
+        let handle = self.inner.loop_handle.clone();
+        self.inner.poll(|lp, io| {
+            let token = try!(lp.add_source(&io));
+            Ok((io, token))
+        }, |io, slot| {
+            Message::Run(Box::new(move || {
+                let res = handle.with_loop(|lp| {
+                    let lp = lp.unwrap();
+                    let token = try!(lp.add_source(&io));
+                    Ok((io, token))
+                });
+                slot.try_produce(res).ok()
+                    .expect("add source try_produce intereference");
+            }))
+        })
     }
 }
 
@@ -1114,38 +1148,6 @@ impl TimeoutState {
     }
 }
 
-impl<E> Source<E> {
-    /// Creates a new `Source` wrapping the provided source of events.
-    pub fn new(e: E) -> Source<E> {
-        Source {
-            readiness: AtomicUsize::new(0),
-            io: e,
-        }
-    }
-}
-
-impl<E: ?Sized> Source<E> {
-    /// Consumes the last readiness notification that this source received.
-    ///
-    /// Currently sources receive readiness notifications on an edge-basis. That
-    /// is, once you receive a notification that an object can be read, you
-    /// won't receive any more notifications until all of that data has been
-    /// read.
-    ///
-    /// The event loop will fill in this information and then inform futures
-    /// that they're ready to go with the `schedule` method, and then the `poll`
-    /// method can use this to figure out what happened.
-    // TODO: shouldn't return a usize here, but rather some kind of newtype
-    pub fn take_readiness(&self) -> usize {
-        self.readiness.swap(0, Ordering::SeqCst)
-    }
-
-    /// Gets access to the underlying I/O object.
-    pub fn io(&self) -> &E {
-        &self.io
-    }
-}
-
 impl Executor for MioSender {
     fn execute_boxed(&self, callback: Box<ExecuteCallback>) {
         self.inner.send(Message::Run(callback))
diff --git a/src/lib.rs b/src/lib.rs
index a54897558..bc665cc7a 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -30,7 +30,7 @@ mod mpsc_queue;
 mod channel;
 
 pub use event_loop::{Loop, LoopPin, LoopHandle, AddSource, AddTimeout};
-pub use event_loop::{LoopData, AddLoopData, TimeoutToken, IoSource, Source};
+pub use event_loop::{LoopData, AddLoopData, TimeoutToken, IoToken};
 pub use readiness_stream::ReadinessStream;
 pub use tcp::{TcpListener, TcpStream};
 pub use timeout::Timeout;
diff --git a/src/readiness_stream.rs b/src/readiness_stream.rs
index 3b48f3131..c67d5cec7 100644
--- a/src/readiness_stream.rs
+++ b/src/readiness_stream.rs
@@ -2,8 +2,9 @@ use std::io;
 use std::sync::atomic::{AtomicUsize, Ordering};
 
 use futures::{Future, Poll};
+use mio;
 
-use event_loop::{IoSource, LoopHandle, AddSource};
+use event_loop::{IoToken, LoopHandle, AddSource};
 
 /// A concrete implementation of a stream of readiness notifications for I/O
 /// objects that originates from an event loop.
@@ -21,31 +22,30 @@ use event_loop::{IoSource, LoopHandle, AddSource};
 /// It's the responsibility of the wrapper to inform the readiness stream when a
 /// "would block" I/O event is seen. The readiness stream will then take care of
 /// any scheduling necessary to get notified when the event is ready again.
-pub struct ReadinessStream {
-    io_token: usize,
-    loop_handle: LoopHandle,
-    source: IoSource,
+pub struct ReadinessStream<E> {
+    token: IoToken,
+    handle: LoopHandle,
     readiness: AtomicUsize,
+    io: E,
 }
 
-pub struct ReadinessStreamNew {
-    inner: AddSource,
-    handle: Option<LoopHandle>,
-    source: Option<IoSource>,
+pub struct ReadinessStreamNew<E> {
+    inner: AddSource<E>,
+    handle: LoopHandle,
 }
 
-impl ReadinessStream {
+impl<E> ReadinessStream<E>
+    where E: mio::Evented + Send + 'static,
+{
     /// Creates a new readiness stream associated with the provided
     /// `loop_handle` and for the given `source`.
     ///
     /// This method returns a future which will resolve to the readiness stream
     /// when it's ready.
-    pub fn new(loop_handle: LoopHandle, source: IoSource)
-               -> ReadinessStreamNew {
+    pub fn new(loop_handle: LoopHandle, source: E) -> ReadinessStreamNew<E> {
         ReadinessStreamNew {
-            inner: loop_handle.add_source(source.clone()),
-            source: Some(source),
-            handle: Some(loop_handle),
+            inner: loop_handle.add_source(source),
+            handle: loop_handle,
         }
     }
 
@@ -60,11 +60,11 @@ impl ReadinessStream {
         if self.readiness.load(Ordering::SeqCst) & 1 != 0 {
             return Poll::Ok(())
         }
-        self.readiness.fetch_or(self.source.take_readiness(), Ordering::SeqCst);
+        self.readiness.fetch_or(self.token.take_readiness(), Ordering::SeqCst);
         if self.readiness.load(Ordering::SeqCst) & 1 != 0 {
             Poll::Ok(())
         } else {
-            self.loop_handle.schedule_read(self.io_token);
+            self.handle.schedule_read(&self.token);
             Poll::NotReady
         }
     }
@@ -80,11 +80,11 @@ impl ReadinessStream {
         if self.readiness.load(Ordering::SeqCst) & 2 != 0 {
             return Poll::Ok(())
         }
-        self.readiness.fetch_or(self.source.take_readiness(), Ordering::SeqCst);
+        self.readiness.fetch_or(self.token.take_readiness(), Ordering::SeqCst);
         if self.readiness.load(Ordering::SeqCst) & 2 != 0 {
             Poll::Ok(())
         } else {
-            self.loop_handle.schedule_write(self.io_token);
+            self.handle.schedule_write(&self.token);
             Poll::NotReady
         }
     }
@@ -102,7 +102,7 @@ impl ReadinessStream {
     /// then again readable.
     pub fn need_read(&self) {
         self.readiness.fetch_and(!1, Ordering::SeqCst);
-        self.loop_handle.schedule_read(self.io_token);
+        self.handle.schedule_read(&self.token);
     }
 
     /// Indicates to this source of events that the corresponding I/O object is
@@ -118,28 +118,48 @@ impl ReadinessStream {
     /// then again writable.
     pub fn need_write(&self) {
         self.readiness.fetch_and(!2, Ordering::SeqCst);
-        self.loop_handle.schedule_write(self.io_token);
+        self.handle.schedule_write(&self.token);
+    }
+
+    /// Returns a reference to the event loop handle that this readiness stream
+    /// is associated with.
+    pub fn loop_handle(&self) -> &LoopHandle {
+        &self.handle
+    }
+
+    /// Returns a shared reference to the underlying I/O object this readiness
+    /// stream is wrapping.
+    pub fn get_ref(&self) -> &E {
+        &self.io
+    }
+
+    /// Returns a mutable reference to the underlying I/O object this readiness
+    /// stream is wrapping.
+    pub fn get_mut(&mut self) -> &mut E {
+        &mut self.io
     }
 }
 
-impl Future for ReadinessStreamNew {
-    type Item = ReadinessStream;
+impl<E> Future for ReadinessStreamNew<E>
+    where E: mio::Evented + Send + 'static,
+{
+    type Item = ReadinessStream<E>;
     type Error = io::Error;
 
-    fn poll(&mut self) -> Poll<ReadinessStream, io::Error> {
-        self.inner.poll().map(|token| {
+    fn poll(&mut self) -> Poll<ReadinessStream<E>, io::Error> {
+        self.inner.poll().map(|(io, token)| {
             ReadinessStream {
-                io_token: token,
-                source: self.source.take().unwrap(),
-                loop_handle: self.handle.take().unwrap(),
+                token: token,
+                handle: self.handle.clone(),
+                io: io,
                 readiness: AtomicUsize::new(0),
             }
         })
     }
 }
 
-impl Drop for ReadinessStream {
+impl<E> Drop for ReadinessStream<E> {
     fn drop(&mut self) {
-        self.loop_handle.drop_source(self.io_token)
+        self.handle.drop_source(&self.token)
     }
 }
diff --git a/src/tcp.rs b/src/tcp.rs
index 5993ef7f0..c0aca32b8 100644
--- a/src/tcp.rs
+++ b/src/tcp.rs
@@ -2,7 +2,6 @@ use std::fmt;
 use std::io::{self, ErrorKind, Read, Write};
 use std::mem;
 use std::net::{self, SocketAddr, Shutdown};
-use std::sync::Arc;
 
 use futures::stream::Stream;
 use futures::{Future, IntoFuture, failed, Poll};
@@ -10,27 +9,21 @@ use futures_io::{IoFuture, IoStream};
 use mio;
 
 use {ReadinessStream, LoopHandle};
-use event_loop::Source;
 
 /// An I/O object representing a TCP socket listening for incoming connections.
 ///
 /// This object can be converted into a stream of incoming connections for
 /// various forms of processing.
 pub struct TcpListener {
-    loop_handle: LoopHandle,
-    ready: ReadinessStream,
-    listener: Arc<Source<mio::tcp::TcpListener>>,
+    io: ReadinessStream<mio::tcp::TcpListener>,
 }
 
 impl TcpListener {
     fn new(listener: mio::tcp::TcpListener,
            handle: LoopHandle) -> IoFuture<TcpListener> {
-        let listener = Arc::new(Source::new(listener));
-        ReadinessStream::new(handle.clone(), listener.clone()).map(|r| {
+        ReadinessStream::new(handle, listener).map(|io| {
             TcpListener {
-                loop_handle: handle,
-                ready: r,
-                listener: listener,
+                io: io,
             }
         }).boxed()
     }
@@ -73,7 +66,7 @@ impl TcpListener {
 
     /// Test whether this socket is ready to be read or not.
     pub fn poll_read(&self) -> Poll<(), io::Error> {
-        self.ready.poll_read()
+        self.io.poll_read()
     }
 
     /// Returns the local address that this listener is bound to.
@@ -81,7 +74,7 @@ impl TcpListener {
     /// This can be useful, for example, when binding to port 0 to figure out
     /// which port was actually bound.
     pub fn local_addr(&self) -> io::Result<SocketAddr> {
-        self.listener.io().local_addr()
+        self.io.get_ref().local_addr()
     }
 
     /// Consumes this listener, returning a stream of the sockets this listener
@@ -99,14 +92,14 @@ impl TcpListener {
             type Error = io::Error;
 
             fn poll(&mut self) -> Poll<Option<Self::Item>, io::Error> {
-                match self.inner.listener.io().accept() {
+                match self.inner.io.get_ref().accept() {
                     Ok(Some(pair)) => {
                         debug!("accepted a socket");
                         Poll::Ok(Some(pair))
                     }
                     Ok(None) => {
                         debug!("waiting to accept another socket");
-                        self.inner.ready.need_read();
+                        self.inner.io.need_read();
                         Poll::NotReady
                     }
                     Err(e) => Poll::Err(e),
@@ -114,17 +107,11 @@ impl TcpListener {
             }
         }
 
-        let loop_handle = self.loop_handle.clone();
+        let loop_handle = self.io.loop_handle().clone();
         Incoming { inner: self }
             .and_then(move |(tcp, addr)| {
-                let tcp = Arc::new(Source::new(tcp));
-                ReadinessStream::new(loop_handle.clone(),
-                                     tcp.clone()).map(move |ready| {
-                    let stream = TcpStream {
-                        source: tcp,
-                        ready: ready,
-                    };
-                    (stream, addr)
+                ReadinessStream::new(loop_handle.clone(), tcp).map(move |io| {
+                    (TcpStream { io: io }, addr)
                 })
             }).boxed()
     }
@@ -132,7 +119,7 @@ impl TcpListener {
 
 impl fmt::Debug for TcpListener {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        self.listener.io().fmt(f)
+        self.io.get_ref().fmt(f)
     }
 }
 
@@ -143,8 +130,7 @@ impl fmt::Debug for TcpListener {
 /// raw underlying I/O object as well as streams for the read/write
 /// notifications on the stream itself.
 pub struct TcpStream {
-    source: Arc<Source<mio::tcp::TcpStream>>,
-    ready: ReadinessStream,
+    io: ReadinessStream<mio::tcp::TcpStream>,
 }
 
 enum TcpStreamNew {
@@ -184,18 +170,8 @@ impl TcpStream {
     fn new(connected_stream: mio::tcp::TcpStream,
            handle: LoopHandle)
            -> IoFuture<TcpStream> {
-        // Once we've connected, wait for the stream to be writable as that's
-        // when the actual connection has been initiated. Once we're writable we
-        // check for `take_socket_error` to see if the connect actually hit an
-        // error or not.
-        //
-        // If all that succeeded then we ship everything on up.
-        let connected_stream = Arc::new(Source::new(connected_stream));
-        ReadinessStream::new(handle, connected_stream.clone()).and_then(|ready| {
-            TcpStreamNew::Waiting(TcpStream {
-                source: connected_stream,
-                ready: ready,
-            })
+        ReadinessStream::new(handle, connected_stream).and_then(|io| {
+            TcpStreamNew::Waiting(TcpStream { io: io })
         }).boxed()
     }
 
@@ -233,7 +209,7 @@ impl TcpStream {
     /// is only suitable for calling in a `Future::poll` method and will
     /// automatically handle ensuring a retry once the socket is readable again.
     pub fn poll_read(&self) -> Poll<(), io::Error> {
-        self.ready.poll_read()
+        self.io.poll_read()
     }
 
     /// Test whether this socket is writey to be written to or not.
@@ -243,17 +219,17 @@ impl TcpStream {
     /// is only suitable for calling in a `Future::poll` method and will
     /// automatically handle ensuring a retry once the socket is writable again.
     pub fn poll_write(&self) -> Poll<(), io::Error> {
-        self.ready.poll_write()
+        self.io.poll_write()
     }
 
     /// Returns the local address that this stream is bound to.
     pub fn local_addr(&self) -> io::Result<SocketAddr> {
-        self.source.io().local_addr()
+        self.io.get_ref().local_addr()
     }
 
     /// Returns the remote address that this stream is connected to.
     pub fn peer_addr(&self) -> io::Result<SocketAddr> {
-        self.source.io().peer_addr()
+        self.io.get_ref().peer_addr()
     }
 
     /// Shuts down the read, write, or both halves of this connection.
@@ -262,7 +238,7 @@ impl TcpStream {
     /// portions to return immediately with an appropriate value (see the
     /// documentation of `Shutdown`).
     pub fn shutdown(&self, how: Shutdown) -> io::Result<()> {
-        self.source.io().shutdown(how)
+        self.io.get_ref().shutdown(how)
     }
 
     /// Sets the value of the `TCP_NODELAY` option on this socket.
@@ -273,12 +249,12 @@ impl TcpStream {
     /// sufficient amount to send out, thereby avoiding the frequent sending of
     /// small packets.
     pub fn set_nodelay(&self, nodelay: bool) -> io::Result<()> {
-        self.source.io().set_nodelay(nodelay)
+        self.io.get_ref().set_nodelay(nodelay)
     }
 
     /// Sets the keepalive time in seconds for this socket.
     pub fn set_keepalive_s(&self, seconds: Option<u32>) -> io::Result<()> {
-        self.source.io().set_keepalive(seconds)
+        self.io.get_ref().set_keepalive(seconds)
     }
 }
 
@@ -291,9 +267,16 @@ impl Future for TcpStreamNew {
             TcpStreamNew::Waiting(s) => s,
             TcpStreamNew::Empty => panic!("can't poll TCP stream twice"),
         };
-        match stream.ready.poll_write() {
+
+        // Once we've connected, wait for the stream to be writable as that's
+        // when the actual connection has been initiated. Once we're writable we
+        // check for `take_socket_error` to see if the connect actually hit an
+        // error or not.
+        //
+        // If all that succeeded then we ship everything on up.
+        match stream.io.poll_write() {
             Poll::Ok(()) => {
-                match stream.source.io().take_socket_error() {
+                match stream.io.get_ref().take_socket_error() {
                     Ok(()) => return Poll::Ok(stream),
                     Err(ref e) if e.kind() == ErrorKind::WouldBlock => {}
                     Err(e) => return Poll::Err(e),
@@ -309,9 +292,9 @@ impl Future for TcpStreamNew {
 
 impl Read for TcpStream {
     fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
-        let r = self.source.io().read(buf);
+        let r = self.io.get_ref().read(buf);
         if is_wouldblock(&r) {
-            self.ready.need_read();
+            self.io.need_read();
         }
         return r
     }
@@ -319,16 +302,16 @@ impl Read for TcpStream {
 
 impl Write for TcpStream {
     fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
-        let r = self.source.io().write(buf);
+        let r = self.io.get_ref().write(buf);
         if is_wouldblock(&r) {
-            self.ready.need_write();
+            self.io.need_write();
         }
         return r
     }
     fn flush(&mut self) -> io::Result<()> {
-        let r = self.source.io().flush();
+        let r = self.io.get_ref().flush();
         if is_wouldblock(&r) {
-            self.ready.need_write();
+            self.io.need_write();
         }
         return r
     }
@@ -336,9 +319,9 @@ impl Write for TcpStream {
 
 impl<'a> Read for &'a TcpStream {
     fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
-        let r = self.source.io().read(buf);
+        let r = self.io.get_ref().read(buf);
         if is_wouldblock(&r) {
-            self.ready.need_read();
+            self.io.need_read();
         }
         return r
     }
@@ -346,17 +329,17 @@ impl<'a> Read for &'a TcpStream {
 
 impl<'a> Write for &'a TcpStream {
     fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
-        let r = self.source.io().write(buf);
+        let r = self.io.get_ref().write(buf);
         if is_wouldblock(&r) {
-            self.ready.need_write();
+            self.io.need_write();
         }
         return r
     }
 
     fn flush(&mut self) -> io::Result<()> {
-        let r = self.source.io().flush();
+        let r = self.io.get_ref().flush();
         if is_wouldblock(&r) {
-            self.ready.need_write();
+            self.io.need_write();
         }
         return r
     }
@@ -371,7 +354,7 @@ fn is_wouldblock<T>(r: &io::Result<T>) -> bool {
 
 impl fmt::Debug for TcpStream {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        self.source.io().fmt(f)
+        self.io.get_ref().fmt(f)
     }
 }
 
@@ -382,13 +365,13 @@ mod sys {
 
     impl AsRawFd for TcpStream {
         fn as_raw_fd(&self) -> RawFd {
-            self.source.io().as_raw_fd()
+            self.io.get_ref().as_raw_fd()
         }
     }
 
     impl AsRawFd for TcpListener {
         fn as_raw_fd(&self) -> RawFd {
-            self.listener.io().as_raw_fd()
+            self.io.get_ref().as_raw_fd()
         }
     }
 }
@@ -402,7 +385,7 @@ mod sys {
     //
     // impl AsRawHandle for TcpStream {
     //     fn as_raw_handle(&self) -> RawHandle {
-    //         self.source.io().as_raw_handle()
+    //         self.io.get_ref().as_raw_handle()
     //     }
     // }
     //
diff --git a/src/udp.rs b/src/udp.rs
index 30c15317d..5a0de2b58 100644
--- a/src/udp.rs
+++ b/src/udp.rs
@@ -1,6 +1,5 @@
 use std::io;
 use std::net::{self, SocketAddr, Ipv4Addr, Ipv6Addr};
-use std::sync::Arc;
 use std::fmt;
 
 use futures::{Future, failed, Poll};
@@ -8,12 +7,10 @@ use futures_io::IoFuture;
 use mio;
 
 use {ReadinessStream, LoopHandle};
-use event_loop::Source;
 
 /// An I/O object representing a UDP socket.
 pub struct UdpSocket {
-    source: Arc<Source<mio::udp::UdpSocket>>,
-    ready: ReadinessStream,
+    io: ReadinessStream<mio::udp::UdpSocket>,
 }
 
 impl LoopHandle {
@@ -34,12 +31,8 @@ impl LoopHandle {
 impl UdpSocket {
     fn new(socket: mio::udp::UdpSocket, handle: LoopHandle)
            -> IoFuture<UdpSocket> {
-        let socket = Arc::new(Source::new(socket));
-        ReadinessStream::new(handle, socket.clone()).map(|ready| {
-            UdpSocket {
-                source: socket,
-                ready: ready,
-            }
+        ReadinessStream::new(handle, socket).map(|io| {
+            UdpSocket { io: io }
         }).boxed()
     }
 
@@ -62,7 +55,7 @@ impl UdpSocket {
 
     /// Returns the local address that this stream is bound to.
     pub fn local_addr(&self) -> io::Result<SocketAddr> {
-        self.source.io().local_addr()
+        self.io.get_ref().local_addr()
     }
 
     /// Test whether this socket is ready to be read or not.
@@ -72,7 +65,7 @@ impl UdpSocket {
     /// is only suitable for calling in a `Future::poll` method and will
     /// automatically handle ensuring a retry once the socket is readable again.
     pub fn poll_read(&self) -> Poll<(), io::Error> {
-        self.ready.poll_read()
+        self.io.poll_read()
     }
 
     /// Test whether this socket is writey to be written to or not.
@@ -82,7 +75,7 @@ impl UdpSocket {
     /// is only suitable for calling in a `Future::poll` method and will
     /// automatically handle ensuring a retry once the socket is writable again.
     pub fn poll_write(&self) -> Poll<(), io::Error> {
-        self.ready.poll_write()
+        self.io.poll_write()
     }
 
     /// Sends data on the socket to the given address. On success, returns the
@@ -91,10 +84,10 @@ impl UdpSocket {
     /// Address type can be any implementor of `ToSocketAddrs` trait. See its
     /// documentation for concrete examples.
     pub fn send_to(&self, buf: &[u8], target: &SocketAddr) -> io::Result<usize> {
-        match self.source.io().send_to(buf, target) {
+        match self.io.get_ref().send_to(buf, target) {
             Ok(Some(n)) => Ok(n),
             Ok(None) => {
-                self.ready.need_write();
+                self.io.need_write();
                 Err(io::Error::new(io::ErrorKind::WouldBlock, "would block"))
             }
             Err(e) => Err(e),
@@ -104,10 +97,10 @@ impl UdpSocket {
     /// Receives data from the socket. On success, returns the number of bytes
     /// read and the address from whence the data came.
     pub fn recv_from(&self, buf: &mut [u8]) -> io::Result<(usize, SocketAddr)> {
-        match self.source.io().recv_from(buf) {
+        match self.io.get_ref().recv_from(buf) {
             Ok(Some(n)) => Ok(n),
             Ok(None) => {
-                self.ready.need_read();
+                self.io.need_read();
                 Err(io::Error::new(io::ErrorKind::WouldBlock, "would block"))
             }
             Err(e) => Err(e),
@@ -121,7 +114,7 @@ impl UdpSocket {
     ///
     /// [link]: #method.set_broadcast
     pub fn broadcast(&self) -> io::Result<bool> {
-        self.source.io().broadcast()
+        self.io.get_ref().broadcast()
     }
 
     /// Sets the value of the `SO_BROADCAST` option for this socket.
@@ -129,7 +122,7 @@ impl UdpSocket {
     /// When enabled, this socket is allowed to send packets to a broadcast
     /// address.
     pub fn set_broadcast(&self, on: bool) -> io::Result<()> {
-        self.source.io().set_broadcast(on)
+        self.io.get_ref().set_broadcast(on)
     }
 
     /// Gets the value of the `IP_MULTICAST_LOOP` option for this socket.
@@ -139,7 +132,7 @@ impl UdpSocket {
     ///
     /// [link]: #method.set_multicast_loop_v4
     pub fn multicast_loop_v4(&self) -> io::Result<bool> {
-        self.source.io().multicast_loop_v4()
+        self.io.get_ref().multicast_loop_v4()
     }
 
     /// Sets the value of the `IP_MULTICAST_LOOP` option for this socket.
@@ -147,7 +140,7 @@ impl UdpSocket {
     /// If enabled, multicast packets will be looped back to the local socket.
     /// Note that this may not have any affect on IPv6 sockets.
     pub fn set_multicast_loop_v4(&self, on: bool) -> io::Result<()> {
-        self.source.io().set_multicast_loop_v4(on)
+        self.io.get_ref().set_multicast_loop_v4(on)
     }
 
     /// Gets the value of the `IP_MULTICAST_TTL` option for this socket.
@@ -157,7 +150,7 @@ impl UdpSocket {
     ///
     /// [link]: #method.set_multicast_ttl_v4
     pub fn multicast_ttl_v4(&self) -> io::Result<u32> {
-        self.source.io().multicast_ttl_v4()
+        self.io.get_ref().multicast_ttl_v4()
     }
 
     /// Sets the value of the `IP_MULTICAST_TTL` option for this socket.
@@ -168,7 +161,7 @@ impl UdpSocket {
     ///
     /// Note that this may not have any affect on IPv6 sockets.
     pub fn set_multicast_ttl_v4(&self, ttl: u32) -> io::Result<()> {
-        self.source.io().set_multicast_ttl_v4(ttl)
+        self.io.get_ref().set_multicast_ttl_v4(ttl)
     }
 
     /// Gets the value of the `IPV6_MULTICAST_LOOP` option for this socket.
@@ -178,7 +171,7 @@ impl UdpSocket {
     ///
     /// [link]: #method.set_multicast_loop_v6
     pub fn multicast_loop_v6(&self) -> io::Result<bool> {
-        self.source.io().multicast_loop_v6()
+        self.io.get_ref().multicast_loop_v6()
     }
 
     /// Sets the value of the `IPV6_MULTICAST_LOOP` option for this socket.
@@ -186,7 +179,7 @@ impl UdpSocket {
     /// Controls whether this socket sees the multicast packets it sends itself.
     /// Note that this may not have any affect on IPv4 sockets.
     pub fn set_multicast_loop_v6(&self, on: bool) -> io::Result<()> {
-        self.source.io().set_multicast_loop_v6(on)
+        self.io.get_ref().set_multicast_loop_v6(on)
     }
 
     /// Gets the value of the `IP_TTL` option for this socket.
@@ -195,7 +188,7 @@ impl UdpSocket {
     ///
     /// [link]: #method.set_ttl
     pub fn ttl(&self) -> io::Result<u32> {
-        self.source.io().ttl()
+        self.io.get_ref().ttl()
     }
 
     /// Sets the value for the `IP_TTL` option on this socket.
@@ -203,7 +196,7 @@ impl UdpSocket {
     /// This value sets the time-to-live field that is used in every packet sent
     /// from this socket.
     pub fn set_ttl(&self, ttl: u32) -> io::Result<()> {
-        self.source.io().set_ttl(ttl)
+        self.io.get_ref().set_ttl(ttl)
     }
 
     /// Executes an operation of the `IP_ADD_MEMBERSHIP` type.
@@ -216,7 +209,7 @@ impl UdpSocket {
     pub fn join_multicast_v4(&self,
                              multiaddr: &Ipv4Addr,
                              interface: &Ipv4Addr) -> io::Result<()> {
-        self.source.io().join_multicast_v4(multiaddr, interface)
+        self.io.get_ref().join_multicast_v4(multiaddr, interface)
     }
 
     /// Executes an operation of the `IPV6_ADD_MEMBERSHIP` type.
@@ -227,7 +220,7 @@ impl UdpSocket {
     pub fn join_multicast_v6(&self,
                              multiaddr: &Ipv6Addr,
                              interface: u32) -> io::Result<()> {
-        self.source.io().join_multicast_v6(multiaddr, interface)
+        self.io.get_ref().join_multicast_v6(multiaddr, interface)
     }
 
     /// Executes an operation of the `IP_DROP_MEMBERSHIP` type.
@@ -239,7 +232,7 @@ impl UdpSocket {
     pub fn leave_multicast_v4(&self,
                               multiaddr: &Ipv4Addr,
                               interface: &Ipv4Addr) -> io::Result<()> {
-        self.source.io().leave_multicast_v4(multiaddr, interface)
+        self.io.get_ref().leave_multicast_v4(multiaddr, interface)
     }
 
     /// Executes an operation of the `IPV6_DROP_MEMBERSHIP` type.
@@ -251,13 +244,13 @@ impl UdpSocket {
     pub fn leave_multicast_v6(&self,
                               multiaddr: &Ipv6Addr,
                               interface: u32) -> io::Result<()> {
-        self.source.io().leave_multicast_v6(multiaddr, interface)
+        self.io.get_ref().leave_multicast_v6(multiaddr, interface)
     }
 }
 
 impl fmt::Debug for UdpSocket {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        self.source.io().fmt(f)
+        self.io.get_ref().fmt(f)
     }
 }
 
@@ -268,7 +261,7 @@ mod sys {
 
     impl AsRawFd for UdpSocket {
         fn as_raw_fd(&self) -> RawFd {
-            self.source.io().as_raw_fd()
+            self.io.get_ref().as_raw_fd()
         }
     }
 }
@@ -282,7 +275,7 @@ mod sys {
     //
     // impl AsRawHandle for UdpSocket {
     //     fn as_raw_handle(&self) -> RawHandle {
-    //         self.source.io().as_raw_handle()
+    //         self.io.get_ref().as_raw_handle()
     //     }
     // }
 }