diff --git a/examples/client.hs b/examples/client.hs
--- a/examples/client.hs
+++ b/examples/client.hs
@@ -19,6 +19,8 @@
                 liftIO $ threadDelay 1000000)
       ["1", "2", "3" :: B.ByteString]
 
+-- | Please run me concurrently with either of the 'server' or
+-- 'serverPipeline' programs.
 main :: IO ()
 main = Z.withContext $ \ctx ->
   runSafeT . runEffect $
diff --git a/examples/server.hs b/examples/server.hs
--- a/examples/server.hs
+++ b/examples/server.hs
@@ -20,6 +20,7 @@
   nxt <- respond (b :| moar)
   echo nxt
 
+-- | Please run me concurrently with the 'client' program.
 main :: IO ()
 main = Z.withContext $ \ctx ->
   runSafeT . runEffect $
diff --git a/examples/serverPipeline.hs b/examples/serverPipeline.hs
new file mode 100644
--- /dev/null
+++ b/examples/serverPipeline.hs
@@ -0,0 +1,26 @@
+{-# OPTIONS_GHC -Wall #-}
+{-# LANGUAGE OverloadedStrings #-}
+
+module Main where
+
+import Control.Monad
+import Pipes.ZMQ4
+import qualified System.ZMQ4 as Z
+
+import qualified Data.ByteString as B
+import Data.List.NonEmpty (NonEmpty(..))
+
+echo :: MonadIO m => Pipe [B.ByteString] (NonEmpty B.ByteString) m ()
+echo = forever $ do
+  nxt <- await
+  case nxt of
+   []          -> liftIO . putStrLn $ "Received empty list."
+   bs@(b:moar) -> do
+     liftIO . putStrLn $ "Received data " ++ show (B.concat bs) ++ "."
+     yield (b :| moar)
+
+-- | Please run me concurrently with the 'client' program.
+main :: IO ()
+main = Z.withContext $ \ctx ->
+  runSafeT . runEffect $
+  setupPipeline ctx Z.Rep (`Z.bind` "ipc:///tmp/server") echo
diff --git a/pipes-zeromq4.cabal b/pipes-zeromq4.cabal
--- a/pipes-zeromq4.cabal
+++ b/pipes-zeromq4.cabal
@@ -1,7 +1,7 @@
 name:                pipes-zeromq4
-version:             0.1.0.0
+version:             0.2.0.0
 synopsis:            Pipes integration for ZeroMQ messaging
-homepage:            http://github.com/peddie/pipes-zeromq4
+homepage:            https://github.com/peddie/pipes-zeromq4
 license:             BSD3
 license-file:        LICENSE
 author:              Matthew Peddie <mpeddie@gmail.com>
@@ -23,7 +23,7 @@
 
 library
   exposed-modules:     Pipes.ZMQ4
-  build-depends:       base >=4.7 && <4.8
+  build-depends:       base >=4.6 && <4.9
                      , zeromq4-haskell >=0.6 && <0.7
                      , bytestring >=0.10 && <0.11
                      , semigroups >=0.16 && <0.17
@@ -32,6 +32,9 @@
   hs-source-dirs:      src
   default-language:    Haskell2010
 
+source-repository head
+  type:           git
+  location:       git://github.com/peddie/pipes-zeromq4
 
 flag examples
     description:    Build usage examples
@@ -79,6 +82,24 @@
     Buildable: False
   hs-source-dirs:      examples
   main-is:             server.hs
+  default-language:    Haskell2010
+  build-depends:       base >= 4.5 && < 5
+                     , pipes-zeromq4
+                     , pipes >=4.1 && <4.2
+                     , pipes-safe >=2.2 && <2.3
+                     , zeromq4-haskell >=0.6 && <0.7
+                     , bytestring >= 0.10 && < 0.11
+                     , semigroups >=0.16 && <0.17
+  ghc-options:         -O2 -rtsopts -threaded
+  ghc-prof-options:    -O2 -prof -fprof-auto -fprof-cafs -rtsopts -threaded
+
+executable serverPipeline
+  if flag(examples)
+    Buildable: True
+  else
+    Buildable: False
+  hs-source-dirs:      examples
+  main-is:             serverPipeline.hs
   default-language:    Haskell2010
   build-depends:       base >= 4.5 && < 5
                      , pipes-zeromq4
diff --git a/src/Pipes/ZMQ4.hs b/src/Pipes/ZMQ4.hs
--- a/src/Pipes/ZMQ4.hs
+++ b/src/Pipes/ZMQ4.hs
@@ -28,6 +28,10 @@
 with exceptions and termination.  If you need to avoid this layer of
 safety, please don't hesitate to contact the author for support.
 
+Currently, everything runs in the same thread as the rest of the
+'Effect', so
+__if something blocks forever on a ZMQ receive or send, you may be in trouble.__
+
 -}
 
 module Pipes.ZMQ4 (
@@ -41,9 +45,14 @@
   -- __Note that according to the ZeroMQ manual pages__, the correct
   -- order of operations is to perform all socket configuration before
   -- running 'Z.connect' or 'Z.bind'.
+
+  -- ** One-directional sources and sinks (for use with "Pipes"'s 'Pipe', 'Producer' and 'Consumer')
   setupProducer
   , setupConsumer
-  , setupBi
+  , setupPipe
+  , setupPipePair
+  , setupPipeline
+    -- ** Bidirectional sources and sinks (for use with "Pipes.Core"'s 'Proxy', 'Server' and 'Client').
   , setupClient
   , setupServer
     -- ** Low-level safe setup
@@ -52,8 +61,8 @@
     -- friends instead of these functions unless you know what you're
     -- doing and need something else.
   , setup
-  , receiveLoop
-  , sendLoop
+  , receiveMessage
+  , sendMessage
     -- * Helpers
   , toNonEmpty
     -- * Re-exported modules
@@ -78,7 +87,7 @@
 -- defined as
 --
 -- @
--- setupProducer ctx ty opts = setup ctx ty opts receiveLoop
+-- setupProducer ctx ty opts = setup ctx ty opts $ forever . receiveMessage
 -- @
 setup :: (MonadSafe m, Base m ~ IO,
           Z.SocketType sockty) =>
@@ -92,25 +101,25 @@
     _ <- liftIO $ opts sock
     act sock
 
--- | This is a low-level function for simply passing all messages
+-- | This is a low-level function for simply passing a message
 -- received on a socket downstream.
 --
 -- @
--- receiveLoop sock = forever $ liftIO (Z.receiveMulti sock) >>= yield
+-- receiveMessage sock = forever $ liftIO (Z.receiveMulti sock) >>= yield
 -- @
-receiveLoop :: (Z.Receiver sck, MonadIO m) =>
-               Z.Socket sck -> Producer [B.ByteString] m ()
-receiveLoop sock = forever $ liftIO (Z.receiveMulti sock) >>= yield
+receiveMessage :: (Z.Receiver sck, MonadIO m) =>
+               Z.Socket sck -> Producer' [B.ByteString] m ()
+receiveMessage sock = liftIO (Z.receiveMulti sock) >>= yield
 
--- | This is a low-level function for simply sending all messages
+-- | This is a low-level function for simply sending a message
 -- available from upstream out on the socket.
 --
 -- @
--- sendLoop sock = forever $ await >>= liftIO . Z.sendMulti sock
+-- sendMessage sock = await >>= liftIO . Z.sendMulti sock
 -- @
-sendLoop :: (Z.Sender sck, MonadIO m) =>
+sendMessage :: (Z.Sender sck, MonadIO m) =>
             Z.Socket sck -> Consumer' (NonEmpty B.ByteString) m ()
-sendLoop sock = forever $ await >>= liftIO . Z.sendMulti sock
+sendMessage sock = await >>= liftIO . Z.sendMulti sock
 
 -- | Create a 'Producer' of message data from the given ZeroMQ
 -- parameters.  All messages received on the socket will be sent
@@ -121,7 +130,7 @@
               -> sockty     -- ^ ZMQ socket type
               -> (Z.Socket sockty -> IO ())  -- ^ Setup function
               -> Producer [B.ByteString] m ()  -- ^ Message source
-setupProducer ctx ty opts = setup ctx ty opts receiveLoop
+setupProducer ctx ty opts = setup ctx ty opts $ forever . receiveMessage
 
 -- | Create a 'Consumer' of message data from the given ZeroMQ
 -- parameters.  All data successfully 'await'ed from upstream will be
@@ -136,33 +145,90 @@
               -> sockty     -- ^ ZMQ socket type
               -> (Z.Socket sockty -> IO ())  -- ^ Setup function
               -> Consumer' (NonEmpty B.ByteString) m ()  -- ^ Message sink
-setupConsumer ctx ty opts = setup ctx ty opts sendLoop
+setupConsumer ctx ty opts = setup ctx ty opts $ forever . sendMessage
 
--- | Create both a 'Producer' and a 'Consumer' of message data, both
--- corresponding to the same socket, from the given ZeroMQ parameters.
--- This is like 'setupProducer' and 'setupConsumer' combined; the
--- socket type must be both a 'Z.Sender' and a 'Z.Receiver' (for
--- example, a 'Z.Dealer').  Messages received over the socket are
--- 'yield'ed by the 'Producer'; messages 'await'ed by the 'Consumer'
--- are sent over the socket.
+-- | Create a 'Pipe' out of a socket from the given ZeroMQ parameters.
+-- All data successfully 'await'ed from upstream will be sent out the
+-- socket, and all message data received on the socket will be
+-- 'yield'ed downstream.
 --
--- See also the descriptions of 'setupProducer' and 'setupConsumer'.
-setupBi :: (MonadSafe m, Base m ~ IO, MonadSafe m1, Base m1 ~ IO,
-            Z.SocketType sockty, Z.Sender sockty, Z.Receiver sockty) =>
-           Z.Context  -- ^ ZMQ context
-        -> sockty     -- ^ ZMQ socket type
-        -> (Z.Socket sockty -> IO ())  -- ^ Setup function
-        -> m1 (Producer [B.ByteString] m (),
-               Consumer (NonEmpty B.ByteString) m ())  -- ^ Message (source, sink) pair
-setupBi ctx ty opts = setup ctx ty opts $ \sock -> return
-  (receiveLoop sock, sendLoop sock)
+-- You can use this 'Pipe' to replace a processing stage in your
+-- pipeline with a ZMQ transaction.  Note that the ZMQ actions here
+-- are blocking; for example, if you make a 'Z.Req' socket, the
+-- pipeline will block when the message from upstream is transmitted
+-- until the response from the remote ZMQ socket is received.
+-- Consider using the @pipes-concurrency@ package if you are sure
+-- you're using the right socket type but you don't want everything to
+-- block here.
+setupPipe :: (MonadSafe m, Base m ~ IO,
+                  Z.SocketType sockty,
+                  Z.Sender sockty, Z.Receiver sockty) =>
+                 Z.Context  -- ^ ZMQ context
+              -> sockty     -- ^ ZMQ socket type
+              -> (Z.Socket sockty -> IO ())  -- ^ Setup function for socket
+              -> Pipe (NonEmpty B.ByteString) [B.ByteString] m ()  -- ^ Pipeline stage via ZMQ
+setupPipe ctx ty opts = setup ctx ty opts $ \sock ->
+  forever $ sendMessage sock >> receiveMessage sock
 
+-- | 'setupPipePair' is the same as 'setupPipe' except that it allows
+-- you to use two different sockets, one for sending and the other for
+-- receiving.
+setupPipePair :: (MonadSafe m, Base m ~ IO,
+                  Z.SocketType txsock, Z.SocketType rxsock,
+                  Z.Sender txsock, Z.Receiver rxsock) =>
+                 Z.Context  -- ^ ZMQ context
+              -> txsock     -- ^ ZMQ transmit socket type
+              -> rxsock     -- ^ ZMQ receive socket type
+              -> (Z.Socket txsock -> IO ())  -- ^ Setup function for transmit socket
+              -> (Z.Socket rxsock -> IO ())  -- ^ Setup function for receive socket
+              -> Pipe (NonEmpty B.ByteString) [B.ByteString] m ()  -- ^ Pipeline stage via ZMQ
+setupPipePair ctx txty rxty txopts rxopts =
+  setup ctx txty txopts $ \tx ->
+  setup ctx rxty rxopts $ \rx ->
+  forever $ sendMessage tx >> receiveMessage rx
+
+-- | Create an 'Effect' out of a socket from the given ZeroMQ
+-- parameters and a pipe to use to connect the resulting 'Producer'
+-- and 'Consumer'.
+--
+-- There are two ways to use a single socket to do both ZMQ sends and
+-- receives in the same pipeline.  'setupPipe' places the ZMQ socket
+-- in the middle of the pipeline.  'setupPipeline' places the same ZMQ
+-- socket at both ends: the provided 'Pipe' is used to connect the
+-- receiving Producer and sending Consumer generated from the ZMQ
+-- socket.  In other words, for @setupPipeline ctx socktype opts
+-- myPipe@,
+--
+-- @
+-- socket RX >-> myPipe >-> socket TX
+-- @
+--
+-- See 'setupPipe' for some caveats.  See @examples/serverPipeline.hs@
+-- for a usage example.
+setupPipeline :: (MonadSafe m, Base m ~ IO,
+                  Z.SocketType sockty,
+                  Z.Sender sockty, Z.Receiver sockty) =>
+                 Z.Context  -- ^ ZMQ context
+              -> sockty     -- ^ ZMQ socket type
+              -> (Z.Socket sockty -> IO ())  -- ^ Setup function for socket
+              -> Pipe [B.ByteString] (NonEmpty B.ByteString) m ()  -- ^ Pipeline to run in the middle
+              -> Effect m ()
+setupPipeline ctx ty opts inpipe = setup ctx ty opts $ \sock ->
+  forever $
+  (liftIO (Z.receiveMulti sock) >>= yield) >->
+  inpipe >->
+  (await >>= liftIO . Z.sendMulti sock)
+
 -- |
 -- This is a low-level function for passing all messages received on
 -- the socket upstream and sending the responses out on the same
 -- socket.
 --
 -- @
+-- clientLoop sock = forever $
+--   liftIO (Z.receiveMulti sock) >>=
+--   request >>=
+--   liftIO . Z.sendMulti sock
 -- @
 clientLoop :: (Z.Receiver sockty, Z.Sender sockty, MonadIO m) =>
               Z.Socket sockty
@@ -189,6 +255,11 @@
 -- socket.
 --
 -- @
+-- serverLoop sock msg =
+--   liftIO (Z.sendMulti sock msg) >>
+--   liftIO (Z.receiveMulti sock) >>=
+--   respond >>=
+--   serverLoop sock
 -- @
 serverLoop :: (Z.Receiver sockty, Z.Sender sockty, MonadIO m) =>
               Z.Socket sockty