diff --git a/README.md b/README.md
--- a/README.md
+++ b/README.md
@@ -40,30 +40,40 @@
 {-# LANGUAGE DerivingStrategies #-}
 {-# LANGUAGE OverloadedStrings #-}
 
-module Main (main) where
+module Ingress (main) where
 
-import Conduit ((.|), awaitForever, runConduit)
 import Control.Monad.IO.Class (MonadIO(liftIO))
 import Data.Binary (Binary)
+import Data.Function ((&))
 import GHC.Generics (Generic)
+import OM.Fork (runRace)
 import OM.Socket (openIngress)
+import Prelude (($), IO, putStrLn)
+import qualified Streaming.Prelude as Stream
 
-{- | The messages that arrive on the socket. -}
+{- |
+  The messages that arrive on the socket.
+
+  This type would typically be shared by both the Ingress and Egress side,
+  or at least the binary encoding must be identical.
+-}
 data Msg
   = A
   | B
   deriving stock (Generic)
   deriving anyclass (Binary)
 
+
 main :: IO ()
 main =
-  runConduit $
+  runRace $
     openIngress "localhost:9000"
-    .| awaitForever (\msg ->
-         case msg of
-           A -> liftIO $ putStrLn "Got A"
-           B -> liftIO $ putStrLn "Got B"
-       )
+    & Stream.mapM_
+       (\msg ->
+           case msg of
+             A -> liftIO $ putStrLn "Got A"
+             B -> liftIO $ putStrLn "Got B"
+         )
 ```
 
   
@@ -75,25 +85,33 @@
 {-# LANGUAGE DerivingStrategies #-}
 {-# LANGUAGE OverloadedStrings #-}
 
-module Main (main) where
+module Egress (main) where
 
-import Conduit ((.|), runConduit, yield)
 import Data.Binary (Binary)
+import Data.Function ((&))
 import GHC.Generics (Generic)
 import OM.Socket (openEgress)
+import Prelude (IO)
+import qualified Streaming.Prelude as Stream
 
-{- | The messages that arrive on the socket. -}
+{- |
+  The messages that arrive on the socket.
+
+  This type would typically be shared between the client and the server,
+  or at least the binary representation must be identical (as, for
+  example, if the client is not written in Haskell).
+-}
 data Msg
   = A
   | B
   deriving stock (Generic)
   deriving anyclass (Binary)
 
+
 main :: IO ()
 main =
-  runConduit $
-    mapM_ yield [A, B, B, A, A, A, B]
-    .| openEgress "localhost:9000"
+  Stream.each [A, B, B, A, A, A, B]
+  & openEgress "localhost:9000"
 
 ```
 
@@ -105,39 +123,55 @@
 {-# LANGUAGE GeneralizedNewtypeDeriving #-}
 {-# LANGUAGE OverloadedStrings #-}
 
-module Main (main) where
 
-import Conduit ((.|), awaitForever, runConduit)
+module Server (main) where
+
 import Control.Monad.Logger (runStdoutLoggingT)
-import Control.Monad.Trans.Class (MonadTrans(lift))
 import Data.Binary (Binary)
+import Data.Function ((&))
+import OM.Fork (runRace)
 import OM.Socket (openServer)
+import Prelude (Maybe(Nothing), ($), IO, Show, String)
+import qualified Streaming.Prelude as Stream
 
-{- | The requests accepted by the server. -}
+
+{-|
+  The requests accepted by the server.
+
+  This type would typically be shared between client and server.
+-}
 newtype Request = EchoRequest String
   deriving newtype (Binary, Show)
 
 
-{- | The response sent back to the client. -}
+{-|
+  The response sent back to the client.
+
+  This type would typically be shared between client and server.
+-}
 newtype Responsee = EchoResponse String
   deriving newtype (Binary, Show)
 
 
-{- | Simple echo resposne server. -}
+{-| Simple echo resposne server. -}
 main :: IO ()
 main =
-  runStdoutLoggingT . runConduit $
-    pure ()
-    .| openServer "localhost:9000" Nothing
-    .| awaitForever (\(EchoRequest str, respond) ->
-        {-
-          You don't necessarily have to respond right away if you don't
-          want to. You can cache the responder away in some state and
-          get back to it at some later time if you like.
-        -}
-        lift $ respond (EchoResponse str)
+  runRace
+  $ runStdoutLoggingT
+  $ (
+      openServer "localhost:9000" Nothing
+      & Stream.mapM_
+          (\ (EchoRequest str, respond) ->
+            {-
+              You don't necessarily have to respond right away if
+              you don't want to. You can cache the responder away in
+              some state and get back to it at some later time if you
+              like. `openServer` and `connectServer` have a mechnamism
+              for handling out of order responses.
+            -}
+            respond (EchoResponse str)
+          )
     )
-
 ```
 
 ### Connect a client to a server
@@ -147,23 +181,32 @@
 {-# LANGUAGE GeneralizedNewtypeDeriving #-}
 {-# LANGUAGE OverloadedStrings #-}
 
-module Main (main) where
+module Client (main) where
 
 import Control.Monad.Logger (runStdoutLoggingT)
 import Data.Binary (Binary)
 import OM.Socket (connectServer)
 
-{- | The requests accepted by the server. -}
+{-|
+  The requests accepted by the server.
+
+  This would typically need to be shared between the client and the
+  server. Certainly the binary encoding must be identical.
+-}
 newtype Request = EchoRequest String
   deriving newtype (Binary, Show)
 
 
-{- | The response sent back to the client. -}
+{-|
+  The response sent back to the client.
+
+  Likewise, this would be shared between client and server.
+-}
 newtype Responsee = EchoResponse String
   deriving newtype (Binary, Show)
 
 
-{- | Simple echo resposne client. -}
+{-| Simple echo resposne client. -}
 main :: IO ()
 main = do
   client <-
diff --git a/examples/Client.hs b/examples/Client.hs
new file mode 100644
--- /dev/null
+++ b/examples/Client.hs
@@ -0,0 +1,39 @@
+{-# LANGUAGE DerivingStrategies #-}
+{-# LANGUAGE GeneralizedNewtypeDeriving #-}
+{-# LANGUAGE OverloadedStrings #-}
+
+module Client (main) where
+
+import Control.Monad.Logger (runStdoutLoggingT)
+import Data.Binary (Binary)
+import OM.Socket (connectServer)
+
+{-|
+  The requests accepted by the server.
+
+  This would typically need to be shared between the client and the
+  server. Certainly the binary encoding must be identical.
+-}
+newtype Request = EchoRequest String
+  deriving newtype (Binary, Show)
+
+
+{-|
+  The response sent back to the client.
+
+  Likewise, this would be shared between client and server.
+-}
+newtype Responsee = EchoResponse String
+  deriving newtype (Binary, Show)
+
+
+{-| Simple echo resposne client. -}
+main :: IO ()
+main = do
+  client <-
+    runStdoutLoggingT $
+      connectServer "localhost:9000" Nothing
+  putStrLn =<< client (EchoRequest "hello")
+  putStrLn =<< client (EchoRequest "world")
+
+
diff --git a/examples/Egress.hs b/examples/Egress.hs
new file mode 100644
--- /dev/null
+++ b/examples/Egress.hs
@@ -0,0 +1,34 @@
+{-# LANGUAGE DeriveAnyClass #-}
+{-# LANGUAGE DeriveGeneric #-}
+{-# LANGUAGE DerivingStrategies #-}
+{-# LANGUAGE OverloadedStrings #-}
+
+module Egress (main) where
+
+import Data.Binary (Binary)
+import Data.Function ((&))
+import GHC.Generics (Generic)
+import OM.Socket (openEgress)
+import Prelude (IO)
+import qualified Streaming.Prelude as Stream
+
+{- |
+  The messages that arrive on the socket.
+
+  This type would typically be shared between the client and the server,
+  or at least the binary representation must be identical (as, for
+  example, if the client is not written in Haskell).
+-}
+data Msg
+  = A
+  | B
+  deriving stock (Generic)
+  deriving anyclass (Binary)
+
+
+main :: IO ()
+main =
+  Stream.each [A, B, B, A, A, A, B]
+  & openEgress "localhost:9000"
+
+
diff --git a/examples/Ingress.hs b/examples/Ingress.hs
new file mode 100644
--- /dev/null
+++ b/examples/Ingress.hs
@@ -0,0 +1,41 @@
+{-# LANGUAGE DeriveAnyClass #-}
+{-# LANGUAGE DeriveGeneric #-}
+{-# LANGUAGE DerivingStrategies #-}
+{-# LANGUAGE OverloadedStrings #-}
+
+module Ingress (main) where
+
+import Control.Monad.IO.Class (MonadIO(liftIO))
+import Data.Binary (Binary)
+import Data.Function ((&))
+import GHC.Generics (Generic)
+import OM.Fork (runRace)
+import OM.Socket (openIngress)
+import Prelude (($), IO, putStrLn)
+import qualified Streaming.Prelude as Stream
+
+{- |
+  The messages that arrive on the socket.
+
+  This type would typically be shared by both the Ingress and Egress side,
+  or at least the binary encoding must be identical.
+-}
+data Msg
+  = A
+  | B
+  deriving stock (Generic)
+  deriving anyclass (Binary)
+
+
+main :: IO ()
+main =
+  runRace $
+    openIngress "localhost:9000"
+    & Stream.mapM_
+       (\msg ->
+           case msg of
+             A -> liftIO $ putStrLn "Got A"
+             B -> liftIO $ putStrLn "Got B"
+         )
+
+
diff --git a/examples/Server.hs b/examples/Server.hs
new file mode 100644
--- /dev/null
+++ b/examples/Server.hs
@@ -0,0 +1,55 @@
+{-# LANGUAGE DerivingStrategies #-}
+{-# LANGUAGE GeneralizedNewtypeDeriving #-}
+{-# LANGUAGE OverloadedStrings #-}
+
+
+module Server (main) where
+
+import Control.Monad.Logger (runStdoutLoggingT)
+import Data.Binary (Binary)
+import Data.Function ((&))
+import OM.Fork (runRace)
+import OM.Socket (openServer)
+import Prelude (Maybe(Nothing), ($), IO, Show, String)
+import qualified Streaming.Prelude as Stream
+
+
+{-|
+  The requests accepted by the server.
+
+  This type would typically be shared between client and server.
+-}
+newtype Request = EchoRequest String
+  deriving newtype (Binary, Show)
+
+
+{-|
+  The response sent back to the client.
+
+  This type would typically be shared between client and server.
+-}
+newtype Responsee = EchoResponse String
+  deriving newtype (Binary, Show)
+
+
+{-| Simple echo resposne server. -}
+main :: IO ()
+main =
+  runRace
+  $ runStdoutLoggingT
+  $ (
+      openServer "localhost:9000" Nothing
+      & Stream.mapM_
+          (\ (EchoRequest str, respond) ->
+            {-
+              You don't necessarily have to respond right away if
+              you don't want to. You can cache the responder away in
+              some state and get back to it at some later time if you
+              like. `openServer` and `connectServer` have a mechnamism
+              for handling out of order responses.
+            -}
+            respond (EchoResponse str)
+          )
+    )
+
+
diff --git a/examples/client.hs b/examples/client.hs
deleted file mode 100644
--- a/examples/client.hs
+++ /dev/null
@@ -1,42 +0,0 @@
-{-# LANGUAGE DerivingStrategies #-}
-{-# LANGUAGE GeneralizedNewtypeDeriving #-}
-{-# LANGUAGE OverloadedStrings #-}
-
-{-# OPTIONS_GHC -Wwarn #-}
-
-module Main (main) where
-
-import Control.Monad.Logger (runStdoutLoggingT)
-import Data.Binary (Binary)
-import OM.Socket (connectServer)
-
-{- | The requests accepted by the server. -}
-newtype Request = EchoRequest String
-  deriving newtype (Binary, Show)
-
-
-{- | The response sent back to the client. -}
-newtype Responsee = EchoResponse String
-  deriving newtype (Binary, Show)
-
-
-{- | Simple echo resposne server. -}
-main :: IO ()
-main = do
-  {-
-    Don't actually call sendRequests, because there is no server running to
-    connect to, which will cause an error, which will cause the test to fail.
-  -}
-  -- sendRequests
-  pure ()
-
-
-sendRequests :: IO ()
-sendRequests = do
-  client <-
-    runStdoutLoggingT $
-      connectServer "localhost:9000" Nothing
-  putStrLn =<< client (EchoRequest "hello")
-  putStrLn =<< client (EchoRequest "world")
-
-
diff --git a/examples/egress.hs b/examples/egress.hs
deleted file mode 100644
--- a/examples/egress.hs
+++ /dev/null
@@ -1,37 +0,0 @@
-{-# LANGUAGE DeriveAnyClass #-}
-{-# LANGUAGE DeriveGeneric #-}
-{-# LANGUAGE DerivingStrategies #-}
-{-# LANGUAGE OverloadedStrings #-}
-
-{-# OPTIONS_GHC -Wwarn #-}
-
-module Main (main) where
-
-import Conduit ((.|), runConduit, yield)
-import Data.Binary (Binary)
-import GHC.Generics (Generic)
-import OM.Socket (openEgress)
-
-{- | The messages that arrive on the socket. -}
-data Msg
-  = A
-  | B
-  deriving stock (Generic)
-  deriving anyclass (Binary)
-
-main :: IO ()
-main =
-  {-
-    Don't actually call sendMessages, because there is no server running to
-    connect to, which will cause an error, which will cause the test to fail.
-  -}
-  -- sendMessages
-  pure ()
-
-sendMessages :: IO ()
-sendMessages = do
-  runConduit $
-    mapM_ yield [A, B, B, A, A, A, B]
-    .| openEgress "localhost:9000"
-
-
diff --git a/examples/ingress.hs b/examples/ingress.hs
deleted file mode 100644
--- a/examples/ingress.hs
+++ /dev/null
@@ -1,42 +0,0 @@
-{-# LANGUAGE DeriveAnyClass #-}
-{-# LANGUAGE DeriveGeneric #-}
-{-# LANGUAGE DerivingStrategies #-}
-{-# LANGUAGE OverloadedStrings #-}
-
-{-# OPTIONS_GHC -Wwarn #-}
-
-module Main (main) where
-
-import Conduit ((.|), awaitForever, runConduit)
-import Control.Monad.IO.Class (MonadIO(liftIO))
-import Data.Binary (Binary)
-import GHC.Generics (Generic)
-import OM.Socket (openIngress)
-
-{- | The messages that arrive on the socket. -}
-data Msg
-  = A
-  | B
-  deriving stock (Generic)
-  deriving anyclass (Binary)
-
-main :: IO ()
-main =
-  {-
-    Don't actually call serveForever, because the "test" we are using to make
-    sure this compiles will never finish running!
-  -}
-  -- serveForever
-  pure ()
-
-serveForever :: IO ()
-serveForever =
-  runConduit $
-    openIngress "localhost:9000"
-    .| awaitForever (\msg ->
-         case msg of
-           A -> liftIO $ putStrLn "Got A"
-           B -> liftIO $ putStrLn "Got B"
-       )
-
-  
diff --git a/examples/server.hs b/examples/server.hs
deleted file mode 100644
--- a/examples/server.hs
+++ /dev/null
@@ -1,49 +0,0 @@
-{-# LANGUAGE DerivingStrategies #-}
-{-# LANGUAGE GeneralizedNewtypeDeriving #-}
-{-# LANGUAGE OverloadedStrings #-}
-
-{-# OPTIONS_GHC -Wwarn #-}
-
-module Main (main) where
-
-import Conduit ((.|), awaitForever, runConduit)
-import Control.Monad.Logger (runStdoutLoggingT)
-import Control.Monad.Trans.Class (MonadTrans(lift))
-import Data.Binary (Binary)
-import OM.Socket (openServer)
-
-{- | The requests accepted by the server. -}
-newtype Request = EchoRequest String
-  deriving newtype (Binary, Show)
-
-
-{- | The response sent back to the client. -}
-newtype Responsee = EchoResponse String
-  deriving newtype (Binary, Show)
-
-
-{- | Simple echo resposne server. -}
-main :: IO ()
-main = do
-  {-
-    Don't actually call server, because the "test" we are using to make
-    sure this compiles will never finish running!
-  -}
-  -- server
-  pure ()
-
-
-server :: IO ()
-server =
-  runStdoutLoggingT . runConduit $
-    pure ()
-    .| openServer "localhost:9000" Nothing
-    .| awaitForever (\(EchoRequest str, respond) ->
-        {-
-          You don't necessarily have to respond right away if you don't
-          want to. You can cache the responder away in some state and
-          get back to it at some later time if you like.
-        -}
-        lift $ respond (EchoResponse str)
-    )
-
diff --git a/om-socket.cabal b/om-socket.cabal
--- a/om-socket.cabal
+++ b/om-socket.cabal
@@ -1,6 +1,6 @@
 cabal-version:       3.0
 name:                om-socket
-version:             0.11.0.5
+version:             1.0.0.0
 synopsis:            Socket utilities.
 description:         Binary ingress server, egress client, and
                      bidirectional binarry client/server
@@ -25,23 +25,24 @@
 
 common dependencies
   build-depends:
-    , aeson          >= 2.0.3.0   && < 2.3
-    , base           >= 4.15.1.0  && < 4.20
-    , binary         >= 0.8.8.0   && < 0.9
-    , binary-conduit >= 1.3.1     && < 1.4
-    , bytestring     >= 0.10.12.1 && < 0.13
-    , conduit        >= 1.3.4.3   && < 1.4
-    , conduit-extra  >= 1.3.6     && < 1.4
-    , containers     >= 0.6.4.1   && < 0.7
-    , exceptions     >= 0.10.4    && < 0.11
-    , megaparsec     >= 9.2.2     && < 9.7
-    , monad-logger   >= 0.3.37    && < 0.4
-    , network        >= 3.1.2.7   && < 3.2
-    , om-show        >= 0.1.2.6   && < 0.2
-    , stm            >= 2.5.0.0   && < 2.6
-    , text           >= 1.2.5.0   && < 2.2
-    , time           >= 1.9.3     && < 1.13
-    , tls            >= 1.5.8     && < 1.10
+    , aeson                >= 2.0.3.0   && < 2.3
+    , base                 >= 4.15.1.0  && < 4.20
+    , binary               >= 0.8.8.0   && < 0.9
+    , bytestring           >= 0.10.12.1 && < 0.13
+    , containers           >= 0.6.4.1   && < 0.7
+    , exceptions           >= 0.10.4    && < 0.11
+    , megaparsec           >= 9.2.2     && < 9.7
+    , monad-logger         >= 0.3.37    && < 0.4
+    , network              >= 3.1.2.7   && < 3.2
+    , om-fork              >= 0.7.1.9   && < 0.8
+    , om-show              >= 0.1.2.6   && < 0.2
+    , stm                  >= 2.5.0.0   && < 2.6
+    , streaming            >= 0.2.4.0   && < 0.3
+    , streaming-binary     >= 0.3.0.1   && < 0.4
+    , streaming-bytestring >= 0.3.2     && < 0.4
+    , text                 >= 1.2.5.0   && < 2.2
+    , tls                  >= 1.5.8     && < 1.10
+    , unliftio-core        >= 0.2.1.0   && < 0.3
 
 library
   import: warnings, dependencies
@@ -52,52 +53,19 @@
   hs-source-dirs: src
   default-language: Haskell2010
 
--- This isn't really a test, it is just used to make sure the example
--- compiles without clusttering up the package with a bunch of
--- executables.
-test-suite ingress-example
-  import: warnings, dependencies
-  main-is: ingress.hs
-  type: exitcode-stdio-1.0
-  hs-source-dirs: examples
-  default-language: Haskell2010
-  build-depends:
-    , om-socket
-
--- This isn't really a test, it is just used to make sure the example
--- compiles without clusttering up the package with a bunch of
--- executables.
-test-suite egress-example
-  import: warnings, dependencies
-  main-is: egress.hs
-  type: exitcode-stdio-1.0
-  hs-source-dirs: examples
-  default-language: Haskell2010
-  build-depends:
-    , om-socket
-
--- This isn't really a test, it is just used to make sure the example
--- compiles without clusttering up the package with a bunch of
--- executables.
-test-suite server-example
-  import: warnings, dependencies
-  main-is: server.hs
-  type: exitcode-stdio-1.0
-  hs-source-dirs: examples
-  default-language: Haskell2010
-  build-depends:
-    , om-socket
-    , transformers >= 0.5.6.2 && < 0.7
-
--- This isn't really a test, it is just used to make sure the example
--- compiles without clusttering up the package with a bunch of
--- executables.
-test-suite client-example
-  import: warnings, dependencies
-  main-is: client.hs
+test-suite test
+  import: dependencies, warnings
+  main-is: test.hs
   type: exitcode-stdio-1.0
-  hs-source-dirs: examples
+  other-modules:
+    Client
+    Egress
+    Ingress
+    Server
+  hs-source-dirs:
+    , test
+    , examples
   default-language: Haskell2010
   build-depends:
     , om-socket
-
+    , hspec >= 2.11.1 && < 2.12
diff --git a/src/OM/Socket.hs b/src/OM/Socket.hs
--- a/src/OM/Socket.hs
+++ b/src/OM/Socket.hs
@@ -4,6 +4,7 @@
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE ScopedTypeVariables #-}
 
 {- | Socket utilities. -}
 module OM.Socket (
@@ -24,30 +25,25 @@
 ) where
 
 
-import Control.Applicative (Alternative((<|>)))
-import Control.Concurrent (Chan, MVar, forkIO, newChan, newEmptyMVar,
-  putMVar, readChan, takeMVar, throwTo, writeChan)
-import Control.Concurrent.STM (TVar, atomically, newTVar, readTVar,
+import Control.Concurrent (MVar, forkIO, newEmptyMVar, putMVar, takeMVar)
+import Control.Concurrent.STM (TVar, atomically, newTVarIO, readTVar,
   retry, writeTVar)
 import Control.Exception (SomeException, bracketOnError, throw)
-import Control.Monad (join, void, when)
+import Control.Monad (when)
 import Control.Monad.Catch (MonadThrow(throwM), MonadCatch, try)
-import Control.Monad.IO.Class (MonadIO(liftIO))
+import Control.Monad.IO.Unlift (MonadUnliftIO)
 import Control.Monad.Logger.CallStack (LoggingT(runLoggingT),
-  MonadLoggerIO(askLoggerIO), logDebug, logError, logWarn)
+  MonadLoggerIO(askLoggerIO), NoLoggingT(runNoLoggingT), MonadLogger,
+  logDebug, logError, logWarn)
 import Data.Aeson (FromJSON, FromJSONKey, ToJSON, ToJSONKey)
 import Data.Binary (Binary(get), encode)
 import Data.Binary.Get (Decoder(Done, Fail, Partial), pushChunk,
   runGetIncremental)
 import Data.ByteString (ByteString)
-import Data.Conduit ((.|), ConduitT, awaitForever, runConduit, transPipe,
-  yield)
-import Data.Conduit.Network (sinkSocket, sourceSocket)
-import Data.Conduit.Serialization.Binary (conduitDecode, conduitEncode)
+import Data.Foldable (traverse_)
 import Data.Map (Map)
 import Data.String (IsString)
 import Data.Text (Text)
-import Data.Time (diffUTCTime, getCurrentTime)
 import Data.Void (Void)
 import Data.Word (Word32)
 import GHC.Generics (Generic)
@@ -60,58 +56,71 @@
 import Network.Socket.ByteString.Lazy (sendAll)
 import Network.TLS (ClientParams, Context, ServerParams, contextNew,
   handshake, recvData, sendData)
+import OM.Fork (Race, race)
 import OM.Show (showt)
 import Prelude (Applicative(pure), Bool(False, True), Bounded(minBound),
   Either(Left, Right), Enum(succ), Eq((/=)), Functor(fmap), Maybe(Just,
-  Nothing), Monad((>>), (>>=), return), MonadFail(fail), Semigroup((<>)),
-  Show(show), ($), (++), (.), (=<<), IO, Monoid, Num, Ord, String,
-  sequence_, snd, userError)
+  Nothing), Monad((>>), (>>=), return), MonadFail(fail), Monoid(mempty),
+  Semigroup((<>)), Show(show), ($), (++), (.), (=<<), IO, Num, Ord,
+  String, flip, snd, userError)
+import Streaming (Alternative((<|>)), MFunctor(hoist), MonadIO(liftIO),
+  MonadTrans(lift), Of, Stream, join, void)
+import Streaming.Binary (decoded)
+import Streaming.ByteString (ByteStream, reread)
 import Text.Megaparsec (MonadParsec(eof), Parsec, many, oneOf, parse,
   satisfy)
 import Text.Megaparsec.Char (char)
 import qualified Data.ByteString as BS
 import qualified Data.ByteString.Lazy as BSL
-import qualified Data.Conduit.List as CL
 import qualified Data.Map as Map
 import qualified Data.Text as T
+import qualified Streaming.Prelude as S
 import qualified Text.Megaparsec as M
 
 
-{- |
-  Opens an "ingress" socket, which is a socket that accepts a stream of
-  messages without responding.
+{-|
+  Opens an "ingress" socket, which is a socket that accepts a stream
+  of messages without responding. In particular, we listen on a socket,
+  accepting new connections, an each connection concurrently reads its
+  elements off the socket and pushes them onto the stream.
 -}
-openIngress :: (Binary i, MonadIO m, MonadFail m)
+openIngress
+  :: forall i m never_returns.
+     ( Binary i
+     , MonadFail m
+     , MonadIO m
+     , Race
+     )
   => AddressDescription
-  -> ConduitT () i m ()
+  -> Stream (Of i) m never_returns
 openIngress bindAddr = do
     so <- listenSocket =<< resolveAddr bindAddr
     mvar <- liftIO newEmptyMVar
-    void . liftIO . forkIO $ acceptLoop so mvar
-    mvarToSource mvar
+    liftIO
+      . runNoLoggingT
+      . race "ingress accept loop"
+      . liftIO
+      $ acceptLoop so mvar
+    mvarToStream mvar
   where
-    mvarToSource :: (MonadIO m) => MVar a -> ConduitT () a m ()
-    mvarToSource mvar = do
-      liftIO (takeMVar mvar) >>= yield
-      mvarToSource mvar
-
-    acceptLoop :: (Binary i) => Socket -> MVar i -> IO ()
+    acceptLoop :: Socket -> MVar i -> IO ()
     acceptLoop so mvar = do
       (conn, _) <- accept so
       void . forkIO $ feed (runGetIncremental get) conn mvar
       acceptLoop so mvar
 
-    feed :: (Binary i) => Decoder i -> Socket -> MVar i -> IO ()
-
+    feed
+      :: Decoder i
+      -> Socket
+      -> MVar i
+      -> IO ()
     feed (Done leftover _ i) conn mvar = do
       putMVar mvar i
       feed (runGetIncremental get `pushChunk` leftover) conn mvar
-
     feed (Partial k) conn mvar = do
       bytes <- recv conn 4096
       when (BS.null bytes) (fail "Socket closed by peer.")
       feed (k (Just bytes)) conn mvar
-
     feed (Fail _ _ err) _conn _chan =
       fail $ "Socket crashed. Decoding error: " ++ show err
 
@@ -124,13 +133,18 @@
   :: ( Binary o
      , MonadFail m
      , MonadIO m
-     , MonadThrow m
      )
   => AddressDescription
-  -> ConduitT o Void m ()
-openEgress addr = do
+  -> Stream (Of o) m r
+  -> m r
+openEgress addr stream = do
   so <- connectSocket =<< resolveAddr addr
-  conduitEncode .| sinkSocket so
+  result <-
+    S.mapM_
+      (liftIO . sendAll so . encode)
+      stream
+  liftIO (close so)
+  pure result
 
 
 {- | Guess the family of a `SockAddr`. -}
@@ -206,118 +220,86 @@
   and provides a way to respond to those requests.
 -}
 openServer
-  :: ( Binary request
+  :: forall request response m never_returns.
+     ( Binary request
      , Binary response
+     , MonadLogger m
+     , MonadCatch m
      , MonadFail m
-     , MonadLoggerIO m
-     , Show request
-     , Show response
+     , MonadUnliftIO m
+     , Race
      )
   => AddressDescription
   -> Maybe (IO ServerParams)
-  -> ConduitT Void (request, response -> m Responded) m ()
+  -> Stream (Of (request, response -> m Responded)) m never_returns
 openServer bindAddr tls = do
     so <- listenSocket =<< resolveAddr bindAddr
-    requestChan <- liftIO newChan
-    logging <- askLoggerIO
-    void . liftIO . forkIO . (`runLoggingT` logging) $ acceptLoop so requestChan
-    chanToSource requestChan
+    requestMVar <- liftIO newEmptyMVar
+    lift
+      . race "server accept loop"
+      $ acceptLoop so requestMVar
+    mvarToStream requestMVar
   where
     acceptLoop
-      :: ( Binary i
-         , Binary o
-         , MonadIO m
-         , MonadLoggerIO n
-         , Show i
-         , Show o
-         )
-      => Socket
-      -> Chan (i, o -> m Responded)
-      -> n ()
-    acceptLoop so requestChan = do
+      :: Socket
+      -> MVar (request, response -> m Responded)
+      -> m void
+    acceptLoop so requestMVar = do
       (conn, ra) <- liftIO (accept so)
-      (inputSource, outputSink) <- prepareConnection conn
       logDebug $ "New connection: " <>  showt ra
-      responseChan <- liftIO newChan
-      logging <- askLoggerIO
-      rtid <-
-        liftIO
-        . forkIO
-        . (`runLoggingT` logging)
-        $ responderThread responseChan outputSink
-      void . liftIO . forkIO . (`runLoggingT` logging) $ do
-        result <- try $ runConduit (
-            pure ()
-            .| transPipe liftIO inputSource
-            .| conduitDecode
-            .| awaitForever (\req@Request {messageId, payload} -> do
-                logDebug $ showt ra <> ": Got request: " <> showt req
-                start <- liftIO getCurrentTime
-                yield (
-                    payload,
-                    \res -> do
-                      liftIO . writeChan responseChan . Response messageId $ res
-                      end <- liftIO getCurrentTime
-                      liftIO . (`runLoggingT` logging) . logDebug
-                        $ showt ra <> ": Responded to " <> showt messageId <> " in ("
-                        <> showt (diffUTCTime end start) <> ")"
-                      pure Responded
-                  )
-              )
-            .| CL.mapM_ (liftIO . writeChan requestChan)
-          )
-        case result of
-          Left err -> liftIO $ throwTo rtid (err :: SomeException)
-          Right () -> return ()
-        logDebug $ "Closed connection: " <>  showt ra
-      acceptLoop so requestChan
+      (input, send) <- prepareConnection conn
+      void . liftIO . forkIO $ handleConnection input send requestMVar
+      acceptLoop so requestMVar
 
-    prepareConnection
-      :: (MonadIO m)
-      => Socket
-      -> m (ConduitT Void ByteString IO (), ConduitT ByteString Void IO ())
-    prepareConnection conn =
-        case tls of
-          Nothing -> pure (sourceSocket conn, sinkSocket conn)
-          Just getParams ->
-            liftIO $ do
-              ctx <- contextNew conn =<< getParams
-              handshake ctx
-              pure (input ctx, output ctx)
+    handleConnection
+      :: ByteStream IO () {-^ raw bytes input from the socket -}
+      -> (BSL.ByteString -> m ()) {-^ How to send bytes back -}
+      -> MVar (request, response -> m Responded)
+         {-^ how we stream (req, respond) tuples to the client code. -}
+      -> IO ()
+    handleConnection input send requestMVar =
+        void $
+          S.mapM_
+            sendRequestToMVar
+            (decoded input)
       where
-        output :: Context -> ConduitT ByteString Void IO ()
-        output ctx = awaitForever (sendData ctx . BSL.fromStrict)
+        sendRequestToMVar :: Request request -> IO ()
+        sendRequestToMVar Request { messageId , payload } =
+          putMVar
+            requestMVar
+            ( payload
+            , respond messageId
+            )
 
-        input :: Context -> ConduitT Void ByteString IO ()
-        input ctx = do
-          bytes <- recvData ctx
-          if BS.null bytes then
-            pure ()
-          else do
-            yield bytes
-            input ctx
+        respond :: MessageId -> response -> m Responded
+        respond responseTo response = do
+          send . encode $ Response { responseTo , response }
+          pure Responded
 
-    responderThread :: (
-          Binary p,
-          MonadLoggerIO m,
-          MonadThrow m,
-          Show p
-        )
-      => Chan (Response p)
-      -> ConduitT ByteString Void IO ()
-      -> m ()
-    responderThread chan outputSink = runConduit (
-        pure ()
-        .| chanToSource chan
-        .| awaitForever (\res@Response {responseTo, response} -> do
-            logDebug
-              $ "Responding to " <> showt responseTo
-              <> " with: " <> showt response
-            yield res
+    {- Maybe make a TLS connection. -}
+    prepareConnection
+      :: Socket
+      -> m
+          ( ByteStream IO ()
+          , BSL.ByteString -> m ()
           )
-        .| conduitEncode
-        .| transPipe liftIO outputSink
-      )
+    prepareConnection conn =
+      case tls of
+        Nothing ->
+          pure
+            ( rereadNull
+                (flip recv 4096)
+                conn
+            , liftIO . sendAll conn
+            )
+        Just getParams ->
+          liftIO $ do
+            ctx <- contextNew conn =<< getParams
+            handshake ctx
+            pure
+              ( rereadNull recvData ctx
+              , sendData ctx
+              )
 
 
 {- |
@@ -326,7 +308,8 @@
   the server.
 -}
 connectServer
-  :: ( Binary request
+  :: forall n request m response.
+     ( Binary request
      , Binary response
      , MonadIO m
      , MonadLoggerIO n
@@ -339,24 +322,26 @@
     logging <- askLoggerIO
     liftIO $ do
       so <- connectSocket =<< resolveAddr addr
-      state <- atomically (newTVar ClientState {
-          csAlive = True,
-          csResponders = Map.empty,
-          csMessageId = minBound,
-          csMessageQueue = []
-        })
-      (send, reqSource) <- prepareConnection so
+      state <-
+        newTVarIO
+          ClientState
+            { csServerAlive = True
+            , csResponders = Map.empty
+            , csMessageId = minBound
+            , csRequestQueue = []
+            }
+      (send, responseSource) <- prepareConnection so
       void . forkIO $ (`runLoggingT` logging) (requestThread send state)
-      void . forkIO $ (`runLoggingT` logging) (responseThread reqSource state)
+      void . forkIO $ (`runLoggingT` logging) (responseThread responseSource state)
       return (\i -> liftIO $ do
           mvar <- newEmptyMVar
           join . atomically $
             readTVar state >>= \case
-              ClientState {csAlive = False} -> return $
+              ClientState {csServerAlive = False} -> return $
                 throwM (userError "Server connection died.")
-              s@ClientState {csMessageQueue} -> do
+              s@ClientState {csRequestQueue} -> do
                 writeTVar state s {
-                    csMessageQueue = csMessageQueue <> [(i, putMVar mvar)]
+                    csRequestQueue = csRequestQueue <> [(i, putMVar mvar)]
                   }
                 return (takeMVar mvar)
         )
@@ -366,51 +351,55 @@
       for TSL or not depending on the configuration.
     -}
     prepareConnection
-      :: (MonadIO m, MonadIO f)
-      => Socket
-      -> f (BSL.ByteString -> IO (), ConduitT Void ByteString m ())
+      :: Socket
+      -> IO
+           ( BSL.ByteString -> IO ()
+           , ByteStream IO ()
+           )
     prepareConnection so =
         case tls of
-          Nothing -> pure (sendAll so, sourceSocket so)
+          Nothing ->
+            pure
+              ( sendAll so
+              , rereadNull
+                  (flip recv 4096)
+                  so
+              )
           Just params -> do
             ctx <- contextNew so params
             handshake ctx
-            pure (send ctx, reqSource ctx)
+            pure (send ctx, resSource ctx)
       where
         send :: Context -> BSL.ByteString -> IO ()
         send = sendData
 
-        reqSource :: (MonadIO m) => Context -> ConduitT Void ByteString m ()
-        reqSource ctx = do
-          bytes <- recvData ctx
-          if BS.null bytes
-            then pure ()
-            else do
-              yield bytes
-              reqSource ctx
+        resSource
+          :: Context
+          -> ByteStream IO ()
+        resSource = do
+          rereadNull recvData
 
-    {- | Receive requests and send them to the server. -}
-    requestThread :: (
-          Binary i,
-          MonadCatch m,
-          MonadLoggerIO m
-        )
-      => (BSL.ByteString -> IO ())
-      -> TVar (ClientState i o)
-      -> m ()
+    {- |
+      Receive requests from the client request function and send them
+      to the server.
+    -}
+    requestThread
+      :: (BSL.ByteString -> IO ())
+      -> TVar (ClientState request response)
+      -> LoggingT IO ()
     requestThread send state =
       join . liftIO . atomically $
         readTVar state >>= \case
-          ClientState {csAlive = False} -> pure (pure ())
-          ClientState {csMessageQueue = []} -> retry
+          ClientState {csServerAlive = False} -> pure (pure ())
+          ClientState {csRequestQueue = []} -> retry
           s@ClientState {
-                csMessageQueue = (m, r):remaining,
+                csRequestQueue = (m, r):remaining,
                 csResponders,
                 csMessageId
               }
             -> do
               writeTVar state s {
-                  csMessageQueue = remaining,
+                  csRequestQueue = remaining,
                   csResponders = Map.insert csMessageId r csResponders,
                   csMessageId = succ csMessageId
                 }
@@ -420,47 +409,64 @@
 
     {- |
       Receive responses from the server and send then them back to the
-      client responder.
+      client request function.
     -}
-    responseThread :: (
-          Binary o,
-          MonadLoggerIO m,
-          MonadCatch m,
-          Show o
-        )
-      => ConduitT Void ByteString m ()
-      -> TVar (ClientState i o)
-      -> m ()
-    responseThread reqSource state = do
-      (try . runConduit) (
-          pure ()
-          .| reqSource
-          .| conduitDecode
-          .| CL.mapM_ (\r@Response {responseTo, response} ->
-              join . liftIO . atomically $
-                readTVar state >>= \ ClientState {csResponders} ->
-                  case Map.lookup responseTo csResponders of
-                    Nothing -> return $
-                      logWarn $ "Unexpected server response: " <> showt r
-                    Just respond -> return $ liftIO (respond response)
+    responseThread
+      :: ByteStream IO ()
+      -> TVar (ClientState request response)
+      -> LoggingT IO ()
+    responseThread resSource stateT = do
+        try
+            (
+              void $
+                S.mapM_
+                  handleResponse
+                  (hoist liftIO (decoded resSource))
             )
-        ) >>= \case
-          Left err ->
-            logError
-              $ "Socket receive error: "
-              <> showt (err :: SomeException)
-          Right () -> return ()
-      join . liftIO . atomically $
-        readTVar state >>= \s@ClientState {csResponders, csMessageQueue} -> do
-          writeTVar state s {csAlive = False}
-          return . liftIO . sequence_ $ [
-              r (throw (userError "Remote connection died."))
-              | r <-
-                  fmap snd (Map.toList csResponders)
-                  <> fmap snd csMessageQueue
-            ]
+          >>= \case
+            Left err -> do
+              logError
+                $ "Socket receive error: "
+                <> showt (err :: SomeException)
+              throw err
+            Right () ->
+              pure ()
+        closeClientState
+      where
+        closeClientState =
+          join . liftIO . atomically $ do
+            state <- readTVar stateT
+            writeTVar stateT state
+              { csServerAlive = False
+              , csResponders = mempty
+              , csRequestQueue = mempty
+              }
 
+            pure . traverse_ liftIO $
+              [ respond (throw (userError "Remote connection died."))
+              | respond <-
+                  Map.elems (csResponders state)
+                  <>  fmap snd (csRequestQueue state)
+              ]
 
+        handleResponse :: Response response -> LoggingT IO ()
+        handleResponse
+            responsePackage@Response
+              { responseTo
+              , response
+              }
+          = do
+              join . lift . atomically $ do
+                state <- readTVar stateT
+                case deleteFind responseTo (csResponders state) of
+                  Nothing ->
+                    pure . logWarn $
+                      "Unexpected server response: " <> showt responsePackage
+                  Just (respond, newResponders) -> do
+                    writeTVar stateT state {csResponders = newResponders}
+                    pure . lift $ respond response
+
+
 {- | A server endpoint configuration. -}
 data Endpoint = Endpoint {
     bindAddr :: AddressDescription,
@@ -470,9 +476,9 @@
 
 
 {- | Response to a request. -}
-data Response p = Response {
-    responseTo :: MessageId,
-      response :: p
+data Response p = Response
+  { responseTo :: MessageId
+  ,   response :: p
   }
   deriving stock (Generic, Show)
 instance (Binary p) => Binary (Response p)
@@ -509,18 +515,18 @@
 
 
 {- | Client connection state. -}
-data ClientState i o = ClientState {
-    csAlive :: Bool,
-    csResponders :: Map MessageId (o -> IO ()),
-    csMessageId :: MessageId,
-    csMessageQueue :: [(i, o -> IO ())]
+data ClientState i o = ClientState
+  {  csServerAlive :: Bool
+  ,   csResponders :: Map MessageId (o -> IO ())
+  ,    csMessageId :: MessageId
+  , csRequestQueue :: [(i, o -> IO ())]
   }
 
 
 {- | A Request message type. -}
-data Request p = Request {
-    messageId :: MessageId,
-      payload :: p
+data Request p = Request
+  { messageId :: MessageId
+  ,   payload :: p
   }
   deriving stock (Generic, Show)
 instance (Binary p) => Binary (Request p)
@@ -533,13 +539,6 @@
   deriving newtype (Binary, Num, Bounded, Eq, Ord, Show, Enum)
 
 
-{- | Construct a coundiut source by reading forever from a 'Chan'. -}
-chanToSource :: (MonadIO m) => Chan a -> ConduitT Void a m ()
-chanToSource chan = do
-  yield =<< liftIO (readChan chan)
-  chanToSource chan
-
-
 {- |
   Proof that a response function was called on the server. Mainly
   useful for including in a type signature somewhere in your server
@@ -547,5 +546,43 @@
   request in all cases.
 -}
 data Responded = Responded
+
+
+mvarToStream
+  :: (MonadIO m)
+  => MVar i
+  -> Stream (Of i) m never_returns
+mvarToStream mvar = do
+  liftIO (takeMVar mvar) >>= S.yield
+  mvarToStream mvar
+
+
+rereadNull
+  :: (Monad m)
+  => (c -> m ByteString)
+  -> c
+  -> ByteStream m ()
+rereadNull f =
+  reread
+    (\c -> do
+      bytes <- f c
+      pure $ if BS.null bytes then Nothing else Just bytes
+    )
+
+
+{-|
+  If the key exists in the map, delete it and return its value along
+  with the new map.
+-}
+deleteFind
+  :: (Ord k)
+  => k
+  -> Map k v
+  -> Maybe (v, Map k v)
+deleteFind key m =
+  case Map.lookup key m of
+    Nothing -> Nothing
+    Just v ->
+      Just (v, Map.delete key m)
 
 
diff --git a/test/test.hs b/test/test.hs
new file mode 100644
--- /dev/null
+++ b/test/test.hs
@@ -0,0 +1,125 @@
+{-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE NumericUnderscores #-}
+{-# LANGUAGE OverloadedStrings #-}
+
+module Main (main) where
+
+import Control.Concurrent (forkIO, newEmptyMVar, putMVar, takeMVar,
+  threadDelay)
+import Control.Monad (void)
+import Control.Monad.Catch (MonadCatch)
+import Control.Monad.IO.Unlift (MonadIO(liftIO), MonadUnliftIO)
+import Control.Monad.Logger (MonadLogger, runStdoutLoggingT)
+import Data.Function ((&))
+import Data.Text (Text)
+import OM.Fork (Race, runRace)
+import OM.Show (showt)
+import OM.Socket (Responded, connectServer, openEgress, openIngress,
+  openServer)
+import Prelude (Applicative(pure), Foldable(length), Functor(fmap),
+  Maybe(Nothing), Traversable(sequence), ($), (.), (<$>), (=<<), IO,
+  Int, MonadFail, fst, print, reverse, seq, sequence_)
+import Streaming.Prelude (Of, Stream)
+import Test.Hspec (hspec, it, shouldBe)
+import qualified Client
+import qualified Egress
+import qualified Ingress
+import qualified Server
+import qualified Streaming.Prelude as Stream
+
+
+main :: IO ()
+main =
+  hspec $ do
+    it "ingress/egress" $ do
+      let
+        expected :: [Int]
+        expected = [1, 2, 3, 4]
+      block <- newEmptyMVar
+      void . forkIO $ do {- Ingress thread -}
+        let
+          instream :: (Race) => Stream (Of Int) IO ()
+          instream = openIngress "localhost:9933"
+        {-
+          The instream never closes, because an ingress server is meant
+          to listen forever, so we consume only the expected number
+          of items from the stream
+        -}
+        actual <-
+          runRace $
+            Stream.take (length expected) instream & Stream.toList_
+        putMVar block actual
+
+      {- Push the expected items over the network.  -}
+      Stream.each expected & openEgress "localhost:9933"
+      actual <- takeMVar block
+      print actual
+      actual `shouldBe` expected
+
+    it "client/server" $ do
+      let
+        requestStream
+          :: ( MonadCatch m
+             , MonadFail m
+             , MonadLogger m
+             , MonadUnliftIO m
+             , Race
+             )
+          => Stream (Of (Int, Text -> m Responded)) m void
+        requestStream = openServer "localhost:9934" Nothing
+
+        requests :: [Int]
+        requests = [1..4]
+
+        expected :: [Text]
+        expected = showt <$> requests
+
+      block <- newEmptyMVar
+      void . forkIO $ runRace $ runStdoutLoggingT $ do
+        {-
+          Handle the requests in reverse order to test out of order
+          handling.
+        -}
+        reqs <-
+          fmap reverse
+          . Stream.toList_
+          . Stream.take (length expected)
+          $ requestStream
+        liftIO (print (fst <$> reqs))
+        sequence_
+          [ respond (showt request)
+          | (request, respond) <- reqs
+          ]
+        liftIO (putMVar block ())
+      call <- runStdoutLoggingT (connectServer "localhost:9934" Nothing)
+      calls <-
+        sequence
+          [ do
+              threadDelay 100_000
+              response <- newEmptyMVar
+              void . forkIO $
+                putMVar response =<< call i
+              pure response
+          | i <- requests
+          ]
+      responses <-
+        sequence
+          [ takeMVar response
+          | response <- calls
+          ]
+      print responses
+      responses `shouldBe` expected
+      takeMVar block
+
+    it "make sure the examples compile" $
+      {-
+        We don't actually _run_ the examples. Really it is the imports
+        that makes sure the examples compile. We `seq` them here just
+        so we don't get warnings about unused imports.
+      -}
+      Client.main
+      `seq` Server.main
+      `seq` Ingress.main
+      `seq` Egress.main
+      `seq` (pure () :: IO ())
+