diff --git a/io-streams.cabal b/io-streams.cabal
--- a/io-streams.cabal
+++ b/io-streams.cabal
@@ -1,5 +1,5 @@
 Name:                io-streams
-Version:             1.1.1.0
+Version:             1.1.2.0
 License:             BSD3
 License-file:        LICENSE
 Category:            Data, Network, IO-Streams
@@ -80,6 +80,10 @@
   .
   /ChangeLog/
   .
+    [@1.1.2.0@] Added @System.IO.Streams.Concurrent.makeChanPipe@, to create a
+                simple concurrent pipe between an @InputStream@/@OutputStream@
+                pair.
+  .
     [@1.1.1.0@] Added @System.IO.Streams.Network.socketToStreamsWithBufferSize@,
                 allowing control over the size of the receive buffers used when
                 reading from sockets.
@@ -188,6 +192,7 @@
                      System.IO.Streams.Tests.ByteString,
                      System.IO.Streams.Tests.Combinators,
                      System.IO.Streams.Tests.Common,
+                     System.IO.Streams.Tests.Concurrent,
                      System.IO.Streams.Tests.Debug,
                      System.IO.Streams.Tests.File,
                      System.IO.Streams.Tests.Handle,
diff --git a/src/System/IO/Streams/Concurrent.hs b/src/System/IO/Streams/Concurrent.hs
--- a/src/System/IO/Streams/Concurrent.hs
+++ b/src/System/IO/Streams/Concurrent.hs
@@ -8,11 +8,14 @@
  , chanToInput
  , chanToOutput
  , concurrentMerge
+ , makeChanPipe
  ) where
 
 ------------------------------------------------------------------------------
+import           Control.Applicative        ((<$>), (<*>))
 import           Control.Concurrent         (forkIO)
-import           Control.Concurrent.Chan    (Chan, readChan, writeChan)
+import           Control.Concurrent.Chan    (Chan, newChan, readChan,
+                                             writeChan)
 import           Control.Concurrent.MVar    (modifyMVar, newEmptyMVar,
                                              newMVar, putMVar, takeMVar)
 import           Control.Exception          (SomeException, mask, throwIO,
@@ -33,7 +36,7 @@
     go = do
         mb <- read is
         writeChan ch mb
-        maybe (return ()) (const go) mb
+        maybe (return $! ()) (const go) mb
 
 
 ------------------------------------------------------------------------------
@@ -80,12 +83,22 @@
         emb <- takeMVar mv
         case emb of
             Left exc      -> throwIO exc
-            Right Nothing -> do b <- modifyMVar nleft $ \n ->
+            Right Nothing -> do x <- modifyMVar nleft $ \n ->
                                      let !n' = n - 1
-                                     in return $! if n' == 0
-                                                    then (n', False)
-                                                    else (n', True)
-                                if b
+                                     in return $! (n', n')
+                                if x > 0
                                   then chunk mv nleft
                                   else return Nothing
             Right x       -> return x
+
+
+--------------------------------------------------------------------------------
+-- | Create a new pair of streams using an underlying 'Chan'. Everything written
+-- to the 'OutputStream' will appear as-is on the 'InputStream'.
+--
+-- Since reading from the 'InputStream' and writing to the 'OutputStream' are
+-- blocking calls, be sure to do so in different threads.
+makeChanPipe :: IO (InputStream a, OutputStream a)
+makeChanPipe = do
+    chan <- newChan
+    (,) <$> chanToInput chan <*> chanToOutput chan
diff --git a/test/System/IO/Streams/Tests/Concurrent.hs b/test/System/IO/Streams/Tests/Concurrent.hs
new file mode 100644
--- /dev/null
+++ b/test/System/IO/Streams/Tests/Concurrent.hs
@@ -0,0 +1,91 @@
+module System.IO.Streams.Tests.Concurrent (tests) where
+
+------------------------------------------------------------------------------
+import           Control.Concurrent
+import           Control.Monad
+import           Prelude                              hiding (lines, read,
+                                                       takeWhile, unlines,
+                                                       unwords, unwords,
+                                                       words)
+import qualified Prelude
+import qualified System.IO.Streams                    as Streams
+import qualified System.IO.Streams.Concurrent         as Streams
+import           System.IO.Streams.Tests.Common
+import           Test.Framework
+import           Test.Framework.Providers.HUnit
+import           Test.Framework.Providers.QuickCheck2
+import           Test.HUnit                           hiding (Test)
+import           Test.QuickCheck                      hiding (output)
+import           Test.QuickCheck.Monadic
+------------------------------------------------------------------------------
+
+tests :: [Test]
+tests = [ testMakeChanPipe
+        , testConcurrentMerge
+        , testConcurrentMergeException
+        , testInputOutput
+        ]
+
+
+------------------------------------------------------------------------------
+testMakeChanPipe :: Test
+testMakeChanPipe = testProperty "concurrent/makeChanPipe" $
+                   monadicIO $
+                   forAllM arbitrary prop
+  where
+    prop :: [Int] -> PropertyM IO ()
+    prop l = liftQ $ do
+        (is, os) <- Streams.makeChanPipe
+        _        <- forkIO $ Streams.writeList l os >> Streams.write Nothing os
+        Streams.toList is >>= assertEqual "makeChanPipe" l
+
+
+------------------------------------------------------------------------------
+testConcurrentMerge :: Test
+testConcurrentMerge = testCase "concurrent/concurrentMerge" $ do
+    mvars <- replicateM nthreads newEmptyMVar
+    chans <- replicateM nthreads newChan
+    let firstMVar = head mvars
+
+    mapM_ (forkIO . ring) $ zip3 mvars (take nthreads $ drop 1 $ cycle mvars)
+                                 chans
+    inputs <- mapM Streams.chanToInput chans
+    resultMVar <- newEmptyMVar
+    forkIO (Streams.concurrentMerge inputs >>= Streams.toList
+                                           >>= putMVar resultMVar)
+    putMVar firstMVar 0
+    result <- takeMVar resultMVar
+    assertEqual "concurrent merge" [0..10] result
+
+  where
+    maxval   = 10 :: Int
+    nthreads = 4  :: Int
+
+    ring (prev, next, chan) = loop
+      where
+        loop = do x <- takeMVar prev
+                  if x > maxval
+                    then do writeChan chan Nothing
+                            putMVar next x
+                    else do writeChan chan $ Just x
+                            threadDelay 100000
+                            putMVar next $! x + 1
+                            loop
+
+------------------------------------------------------------------------------
+testConcurrentMergeException :: Test
+testConcurrentMergeException =
+    testCase "concurrent/concurrentMerge/exception" $ do
+        inp <- Streams.makeInputStream (error "bad") >>=
+               Streams.concurrentMerge . (:[])
+        expectExceptionH (Streams.toList inp)
+
+
+------------------------------------------------------------------------------
+testInputOutput :: Test
+testInputOutput = testCase "concurrent/input-output" $ do
+    is   <- Streams.fromList [1..10::Int]
+    chan <- newChan
+    is'  <- Streams.chanToInput chan
+    Streams.inputToChan is chan
+    Streams.toList is' >>= assertEqual "input-output" [1..10]
diff --git a/test/TestSuite.hs b/test/TestSuite.hs
--- a/test/TestSuite.hs
+++ b/test/TestSuite.hs
@@ -4,6 +4,7 @@
 import qualified System.IO.Streams.Tests.Builder     as Builder
 import qualified System.IO.Streams.Tests.ByteString  as ByteString
 import qualified System.IO.Streams.Tests.Combinators as Combinators
+import qualified System.IO.Streams.Tests.Concurrent  as Concurrent
 import qualified System.IO.Streams.Tests.Debug       as Debug
 import qualified System.IO.Streams.Tests.File        as File
 import qualified System.IO.Streams.Tests.Handle      as Handle
@@ -26,6 +27,7 @@
             , testGroup "Tests.ByteString" ByteString.tests
             , testGroup "Tests.Debug" Debug.tests
             , testGroup "Tests.Combinators" Combinators.tests
+            , testGroup "Tests.Concurrent" Concurrent.tests
             , testGroup "Tests.File" File.tests
             , testGroup "Tests.Handle" Handle.tests
             , testGroup "Tests.Internal" Internal.tests