From 494bdcbe564c45894bed1854be0fd156c8e1d615 Mon Sep 17 00:00:00 2001
From: Victor Denisov <denisovenator@gmail.com>
Date: Tue, 26 Apr 2016 23:19:30 -0700
Subject: [PATCH 01/14] Add tls implementation

---
 Database/MongoDB/Internal/Tls.hs | 67 ++++++++++++++++++++++++++++++++
 1 file changed, 67 insertions(+)
 create mode 100644 Database/MongoDB/Internal/Tls.hs

diff --git a/Database/MongoDB/Internal/Tls.hs b/Database/MongoDB/Internal/Tls.hs
new file mode 100644
index 0000000..191a1f4
--- /dev/null
+++ b/Database/MongoDB/Internal/Tls.hs
@@ -0,0 +1,67 @@
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE RecordWildCards #-}
+
+-- | TLS connection to mongodb
+
+module Bend.Database.Mongo.Tls
+(
+  connect,
+)
+where
+
+import Data.IORef
+import Data.Monoid
+import Data.Text (Text)
+import qualified Data.Text as Text
+import qualified Data.ByteString as ByteString
+import qualified Data.ByteString.Lazy as Lazy.ByteString
+import Data.Default.Class (def)
+import Control.Exception (bracketOnError)
+import System.IO
+import Database.MongoDB (Pipe)
+import Database.MongoDB.Internal.Protocol (newPipeWith)
+import Database.MongoDB.Internal.Connection (Connection(Connection))
+import qualified Database.MongoDB.Internal.Connection as Connection
+import qualified Network
+import qualified Network.TLS as TLS
+import qualified Network.TLS.Extra.Cipher as TLS
+import qualified Control.IO.Region as Region
+
+-- | Connect to mongodb using TLS
+connect :: Text -> Int -> IO Pipe
+connect host port = bracketOnError Region.open Region.close $ \r -> do
+  handle <- Region.alloc_ r
+    (Network.connectTo (Text.unpack host)
+                       (Network.PortNumber $ fromIntegral port))
+    hClose
+
+  let params = (TLS.defaultParamsClient (Text.unpack host) "")
+        { TLS.clientSupported = def
+            { TLS.supportedCiphers = TLS.ciphersuite_all}
+        , TLS.clientHooks = def
+            { TLS.onServerCertificate = \_ _ _ _ -> return []}
+        }
+  context <- Region.alloc_ r
+    (TLS.contextNew handle params)
+    TLS.contextClose
+  TLS.handshake context
+
+  conn <- tlsConnection context (Region.close r)
+  newPipeWith conn
+
+tlsConnection :: TLS.Context -> IO () -> IO Connection
+tlsConnection ctx close = do
+  restRef <- newIORef mempty
+  return Connection
+    { Connection.read = do
+        rest <- readIORef restRef
+        writeIORef restRef mempty
+        if ByteString.null rest
+          then TLS.recvData ctx
+          else return rest
+    , Connection.unread = \rest ->
+        modifyIORef restRef (rest <>)
+    , Connection.write = TLS.sendData ctx . Lazy.ByteString.fromStrict
+    , Connection.flush = TLS.contextFlush ctx
+    , Connection.close = close
+    }

From b3effd44397a3a7a6eeeb6a35943ac9fb5d1b1af Mon Sep 17 00:00:00 2001
From: Victor Denisov <denisovenator@gmail.com>
Date: Wed, 27 Apr 2016 00:04:33 -0700
Subject: [PATCH 02/14] Increase benchmark size for better resolution

---
 Benchmark.hs  | 4 ++--
 mongoDB.cabal | 3 +++
 2 files changed, 5 insertions(+), 2 deletions(-)

diff --git a/Benchmark.hs b/Benchmark.hs
index a299ad7..c32984a 100644
--- a/Benchmark.hs
+++ b/Benchmark.hs
@@ -11,11 +11,11 @@ import Database.MongoDB.Query
 import qualified Data.Text as T
 
 main = defaultMain [
-    bgroup "insert" [ bench "100" $ nfIO doInserts ]
+    bgroup "insert" [ bench "1000" $ nfIO doInserts ]
   ]
 
 doInserts = do
-    let docs = (flip map) [0..100] $ \i ->
+    let docs = (flip map) [0..1000] $ \i ->
             ["name" M.=: (T.pack $ "name " ++ (show i))]
 
     pipe <- M.connect (M.host "127.0.0.1")
diff --git a/mongoDB.cabal b/mongoDB.cabal
index d8354df..884be35 100644
--- a/mongoDB.cabal
+++ b/mongoDB.cabal
@@ -82,6 +82,8 @@ Benchmark bench
   type:               exitcode-stdio-1.0
   Build-depends:      array -any
                     , base < 5
+                    , base64-bytestring
+                    , base16-bytestring
                     , binary -any
                     , bson >= 0.3 && < 0.4
                     , text
@@ -90,6 +92,7 @@ Benchmark bench
                     , mtl >= 2
                     , cryptohash -any
                     , network -any
+                    , nonce
                     , parsec -any
                     , random -any
                     , random-shuffle -any

From 19e631c9f4621bfbe517e686f9bedc00ec1191fd Mon Sep 17 00:00:00 2001
From: Victor Denisov <denisovenator@gmail.com>
Date: Wed, 27 Apr 2016 00:10:12 -0700
Subject: [PATCH 03/14] Implement readExactly using hGet

---
 Database/MongoDB/Internal/Connection.hs | 37 ++-----------------------
 Database/MongoDB/Internal/Protocol.hs   |  4 +--
 2 files changed, 4 insertions(+), 37 deletions(-)

diff --git a/Database/MongoDB/Internal/Connection.hs b/Database/MongoDB/Internal/Connection.hs
index d921941..c0abdd7 100644
--- a/Database/MongoDB/Internal/Connection.hs
+++ b/Database/MongoDB/Internal/Connection.hs
@@ -23,49 +23,16 @@ import System.IO.Error (mkIOError, eofErrorType)
 --
 -- `read` should return `ByteString.null` on EOF
 data Connection = Connection {
-    read :: IO ByteString,
-    unread :: ByteString -> IO (),
+    readExactly :: Int -> IO ByteString,
     write :: ByteString -> IO (),
     flush :: IO (),
     close :: IO ()}
 
-readExactly :: Connection -> Int -> IO Lazy.ByteString
--- ^ Read specified number of bytes
---
--- If EOF is reached before N bytes then raise EOF exception.
-readExactly conn count = go mempty count
-  where
-  go acc n = do
-    -- read until get enough bytes
-    chunk <- read conn
-    when (ByteString.null chunk) $
-      ioError eof
-    let len = ByteString.length chunk
-    if len >= n
-      then do
-        let (res, rest) = ByteString.splitAt n chunk
-        unless (ByteString.null rest) $
-          unread conn rest
-        return (acc <> Lazy.ByteString.fromStrict res)
-      else go (acc <> Lazy.ByteString.fromStrict chunk) (n - len)
-  eof = mkIOError eofErrorType "Database.MongoDB.Internal.Connection"
-                  Nothing Nothing
-
 fromHandle :: Handle -> IO Connection
 -- ^ Make connection form handle
 fromHandle handle = do
-  restRef <- newIORef mempty
   return Connection
-    { read = do
-        rest <- readIORef restRef
-        writeIORef restRef mempty
-        if ByteString.null rest
-          -- 32k corresponds to the default chunk size
-          -- used in bytestring package
-          then ByteString.hGetSome handle (32 * 1024)
-          else return rest
-    , unread = \rest ->
-        modifyIORef restRef (rest <>)
+    { readExactly = ByteString.hGet handle
     , write = ByteString.hPut handle
     , flush = hFlush handle
     , close = hClose handle
diff --git a/Database/MongoDB/Internal/Protocol.hs b/Database/MongoDB/Internal/Protocol.hs
index 53db3be..1f3b534 100644
--- a/Database/MongoDB/Internal/Protocol.hs
+++ b/Database/MongoDB/Internal/Protocol.hs
@@ -117,8 +117,8 @@ readMessage :: Connection -> IO Response
 -- ^ read response from a connection
 readMessage conn = readResp  where
     readResp = do
-        len <- fromEnum . decodeSize <$> Connection.readExactly conn 4
-        runGet getReply <$> Connection.readExactly conn len
+        len <- fromEnum . decodeSize . L.fromStrict <$> Connection.readExactly conn 4
+        runGet getReply . L.fromStrict <$> Connection.readExactly conn len
     decodeSize = subtract 4 . runGet getInt32
 
 type FullCollection = Text

From f956cb26237508bd04ec6240e383086b416f7fca Mon Sep 17 00:00:00 2001
From: Victor Denisov <denisovenator@gmail.com>
Date: Sat, 30 Apr 2016 20:11:44 -0700
Subject: [PATCH 04/14] Incorporate Tls implementation

---
 Database/MongoDB/Internal/Tls.hs | 38 +++++++++++++-----
 mongoDB.cabal                    |  4 ++
 test/Internal/ConnectionSpec.hs  | 69 --------------------------------
 3 files changed, 33 insertions(+), 78 deletions(-)
 delete mode 100644 test/Internal/ConnectionSpec.hs

diff --git a/Database/MongoDB/Internal/Tls.hs b/Database/MongoDB/Internal/Tls.hs
index 191a1f4..3579a9b 100644
--- a/Database/MongoDB/Internal/Tls.hs
+++ b/Database/MongoDB/Internal/Tls.hs
@@ -3,7 +3,7 @@
 
 -- | TLS connection to mongodb
 
-module Bend.Database.Mongo.Tls
+module Database.MongoDB.Internal.Tls
 (
   connect,
 )
@@ -16,12 +16,15 @@ import qualified Data.Text as Text
 import qualified Data.ByteString as ByteString
 import qualified Data.ByteString.Lazy as Lazy.ByteString
 import Data.Default.Class (def)
+import Control.Applicative ((<$>))
 import Control.Exception (bracketOnError)
+import Control.Monad (when, unless)
 import System.IO
 import Database.MongoDB (Pipe)
 import Database.MongoDB.Internal.Protocol (newPipeWith)
 import Database.MongoDB.Internal.Connection (Connection(Connection))
 import qualified Database.MongoDB.Internal.Connection as Connection
+import System.IO.Error (mkIOError, eofErrorType)
 import qualified Network
 import qualified Network.TLS as TLS
 import qualified Network.TLS.Extra.Cipher as TLS
@@ -53,14 +56,31 @@ tlsConnection :: TLS.Context -> IO () -> IO Connection
 tlsConnection ctx close = do
   restRef <- newIORef mempty
   return Connection
-    { Connection.read = do
-        rest <- readIORef restRef
-        writeIORef restRef mempty
-        if ByteString.null rest
-          then TLS.recvData ctx
-          else return rest
-    , Connection.unread = \rest ->
-        modifyIORef restRef (rest <>)
+    { Connection.readExactly = \count -> let
+          readSome = do
+            rest <- readIORef restRef
+            writeIORef restRef mempty
+            if ByteString.null rest
+              then TLS.recvData ctx
+              else return rest
+          unread = \rest ->
+            modifyIORef restRef (rest <>)
+          go acc n = do
+            -- read until get enough bytes
+            chunk <- readSome
+            when (ByteString.null chunk) $
+              ioError eof
+            let len = ByteString.length chunk
+            if len >= n
+              then do
+                let (res, rest) = ByteString.splitAt n chunk
+                unless (ByteString.null rest) $
+                  unread rest
+                return (acc <> Lazy.ByteString.fromStrict res)
+              else go (acc <> Lazy.ByteString.fromStrict chunk) (n - len)
+          eof = mkIOError eofErrorType "Database.MongoDB.Internal.Connection"
+                Nothing Nothing
+       in Lazy.ByteString.toStrict <$> go mempty count
     , Connection.write = TLS.sendData ctx . Lazy.ByteString.fromStrict
     , Connection.flush = TLS.contextFlush ctx
     , Connection.close = close
diff --git a/mongoDB.cabal b/mongoDB.cabal
index 884be35..e7cc087 100644
--- a/mongoDB.cabal
+++ b/mongoDB.cabal
@@ -34,11 +34,14 @@ Library
                     , mtl >= 2
                     , cryptohash -any
                     , network -any
+                    , io-region -any
                     , parsec -any
                     , random -any
                     , random-shuffle -any
                     , monad-control >= 0.3.1
                     , lifted-base >= 0.1.0.3
+                    , tls >= 1.2.0
+                    , data-default-class -any
                     , transformers-base >= 0.4.1
                     , hashtables >= 1.1.2.0
                     , base16-bytestring >= 0.1.1.6
@@ -49,6 +52,7 @@ Library
                     Database.MongoDB.Admin
                     Database.MongoDB.Connection
                     Database.MongoDB.Internal.Connection
+                    Database.MongoDB.Internal.Tls
                     Database.MongoDB.Internal.Protocol
                     Database.MongoDB.Internal.Util
                     Database.MongoDB.Query
diff --git a/test/Internal/ConnectionSpec.hs b/test/Internal/ConnectionSpec.hs
deleted file mode 100644
index 89ef85b..0000000
--- a/test/Internal/ConnectionSpec.hs
+++ /dev/null
@@ -1,69 +0,0 @@
-
-module Internal.ConnectionSpec (
-  spec,
-) where
-
-import Prelude hiding (read)
-import Data.Monoid
-import Data.IORef
-import Control.Monad
-import System.IO.Error (isEOFError)
-import Test.Hspec
-
-import Database.MongoDB.Internal.Connection
-
-spec :: Spec
-spec = describe "Internal.Connection" $ do
-  readExactlySpec
-
-readExactlySpec :: Spec
-readExactlySpec = describe "readExactly" $ do
-  it "should return specified number of bytes" $ do
-    let conn = Connection
-          { read = return "12345"
-          , unread = \_ -> return ()
-          , write = \_ -> return ()
-          , flush = return ()
-          , close = return ()
-          }
-
-    res <- readExactly conn 3
-    res `shouldBe` "123"
-
-  it "should unread the rest" $ do
-    restRef <- newIORef mempty
-    let conn = Connection
-          { read = return "12345"
-          , unread = writeIORef restRef
-          , write = \_ -> return ()
-          , flush = return ()
-          , close = return ()
-          }
-
-    void $ readExactly conn 3
-    rest <- readIORef restRef
-    rest `shouldBe` "45"
-
-  it "should ask for more bytes if the first chunk is too small" $ do
-    let conn = Connection
-          { read = return "12345"
-          , unread = \_ -> return ()
-          , write = \_ -> return ()
-          , flush = return ()
-          , close = return ()
-          }
-
-    res <- readExactly conn 8
-    res `shouldBe` "12345123"
-
-  it "should throw on EOF" $ do
-    let conn = Connection
-          { read = return mempty
-          , unread = \_ -> return ()
-          , write = \_ -> return ()
-          , flush = return ()
-          , close = return ()
-          }
-
-    void $ readExactly conn 3
-      `shouldThrow` isEOFError

From c011b1a23ceb8574ab41c1407db9ecca7d5b9583 Mon Sep 17 00:00:00 2001
From: Victor Denisov <denisovenator@gmail.com>
Date: Sun, 10 Apr 2016 17:45:30 -0700
Subject: [PATCH 05/14] Move content of System.IO.Pipeline to Internal.Protocol

---
 Database/MongoDB/Connection.hs        |   3 +-
 Database/MongoDB/Internal/Protocol.hs | 112 ++++++++++++++++++++++--
 System/IO/Pipeline.hs                 | 118 --------------------------
 mongoDB.cabal                         |   1 -
 4 files changed, 107 insertions(+), 127 deletions(-)
 delete mode 100644 System/IO/Pipeline.hs

diff --git a/Database/MongoDB/Connection.hs b/Database/MongoDB/Connection.hs
index b0337bc..875603a 100644
--- a/Database/MongoDB/Connection.hs
+++ b/Database/MongoDB/Connection.hs
@@ -42,12 +42,11 @@ import Data.Text (Text)
 import qualified Data.Bson as B
 import qualified Data.Text as T
 
-import Database.MongoDB.Internal.Protocol (Pipe, newPipe)
+import Database.MongoDB.Internal.Protocol (Pipe, newPipe, close, isClosed)
 import Database.MongoDB.Internal.Util (untilSuccess, liftIOE,
                                        updateAssocs, shuffle, mergesortM)
 import Database.MongoDB.Query (Command, Failure(ConnectionFailure), access,
                               slaveOk, runCommand)
-import System.IO.Pipeline (close, isClosed)
 
 adminCommand :: Command -> Pipe -> IO Document
 -- ^ Run command against admin database on server connected to pipe. Fail if connection fails.
diff --git a/Database/MongoDB/Internal/Protocol.hs b/Database/MongoDB/Internal/Protocol.hs
index 1f3b534..8385659 100644
--- a/Database/MongoDB/Internal/Protocol.hs
+++ b/Database/MongoDB/Internal/Protocol.hs
@@ -8,6 +8,14 @@
 {-# LANGUAGE CPP, FlexibleContexts, TupleSections, TypeSynonymInstances #-}
 {-# LANGUAGE MultiParamTypeClasses, FlexibleInstances, UndecidableInstances #-}
 
+{-# LANGUAGE NamedFieldPuns, ScopedTypeVariables #-}
+
+#if (__GLASGOW_HASKELL__ >= 706)
+{-# LANGUAGE RecursiveDo #-}
+#else
+{-# LANGUAGE DoRec #-}
+#endif
+
 module Database.MongoDB.Internal.Protocol (
     FullCollection,
     -- * Pipe
@@ -19,7 +27,8 @@ module Database.MongoDB.Internal.Protocol (
     -- ** Reply
     Reply(..), ResponseFlag(..),
     -- * Authentication
-    Username, Password, Nonce, pwHash, pwKey
+    Username, Password, Nonce, pwHash, pwKey,
+    isClosed, close
 ) where
 
 #if !MIN_VERSION_base(4,8,0)
@@ -35,6 +44,13 @@ import Data.IORef (IORef, newIORef, atomicModifyIORef)
 import System.IO (Handle)
 import System.IO.Unsafe (unsafePerformIO)
 import Data.Maybe (maybeToList)
+import GHC.Conc (ThreadStatus(..), threadStatus)
+import Control.Monad (forever)
+import Control.Concurrent.Chan (Chan, newChan, readChan, writeChan)
+import Control.Concurrent (ThreadId, forkIO, killThread)
+
+import Control.Exception.Lifted (onException, throwIO, try)
+import qualified Control.Exception.Lifted as CEL
 
 import qualified Data.ByteString.Lazy as L
 
@@ -49,13 +65,97 @@ import qualified Data.Text as T
 import qualified Data.Text.Encoding as TE
 
 import Database.MongoDB.Internal.Util (whenJust, bitOr, byteStringHex)
-import System.IO.Pipeline (Pipeline, newPipeline, IOStream(..))
-
-import qualified System.IO.Pipeline as P
+import System.IO (hClose, hFlush)
 
 import Database.MongoDB.Internal.Connection (Connection)
 import qualified Database.MongoDB.Internal.Connection as Connection
 
+#if MIN_VERSION_base(4,6,0)
+import Control.Concurrent.MVar.Lifted (MVar, newEmptyMVar, newMVar, withMVar,
+                                       putMVar, readMVar, mkWeakMVar)
+#else
+import Control.Concurrent.MVar.Lifted (MVar, newEmptyMVar, newMVar, withMVar,
+                                         putMVar, readMVar, addMVarFinalizer)
+#endif
+
+#if !MIN_VERSION_base(4,6,0)
+mkWeakMVar :: MVar a -> IO () -> IO ()
+mkWeakMVar = addMVarFinalizer
+#endif
+
+-- * IOStream
+
+-- | An IO sink and source where value of type @o@ are sent and values of type @i@ are received.
+data IOStream i o = IOStream {
+    writeStream :: o -> IO (),
+    readStream :: IO i,
+    closeStream :: IO () }
+
+-- * Pipeline
+
+-- | Thread-safe and pipelined connection
+data Pipeline i o = Pipeline {
+    vStream :: MVar (IOStream i o),  -- ^ Mutex on handle, so only one thread at a time can write to it
+    responseQueue :: Chan (MVar (Either IOError i)),  -- ^ Queue of threads waiting for responses. Every time a response arrive we pop the next thread and give it the response.
+    listenThread :: ThreadId
+    }
+
+-- | Create new Pipeline over given handle. You should 'close' pipeline when finished, which will also close handle. If pipeline is not closed but eventually garbage collected, it will be closed along with handle.
+newPipeline :: IOStream i o -> IO (Pipeline i o)
+newPipeline stream = do
+    vStream <- newMVar stream
+    responseQueue <- newChan
+    rec
+        let pipe = Pipeline{..}
+        listenThread <- forkIO (listen pipe)
+    _ <- mkWeakMVar vStream $ do
+        killThread listenThread
+        closeStream stream
+    return pipe
+
+close :: Pipeline i o -> IO ()
+-- ^ Close pipe and underlying connection
+close Pipeline{..} = do
+    killThread listenThread
+    closeStream =<< readMVar vStream
+
+isClosed :: Pipeline i o -> IO Bool
+isClosed Pipeline{listenThread} = do
+    status <- threadStatus listenThread
+    return $ case status of
+        ThreadRunning -> False
+        ThreadFinished -> True
+        ThreadBlocked _ -> False
+        ThreadDied -> True
+--isPipeClosed Pipeline{..} = isClosed =<< readMVar vHandle  -- isClosed hangs while listen loop is waiting on read
+
+listen :: Pipeline i o -> IO ()
+-- ^ Listen for responses and supply them to waiting threads in order
+listen Pipeline{..} = do
+    stream <- readMVar vStream
+    forever $ do
+        e <- try $ readStream stream
+        var <- readChan responseQueue
+        putMVar var e
+        case e of
+            Left err -> closeStream stream >> ioError err  -- close and stop looping
+            Right _ -> return ()
+
+psend :: Pipeline i o -> o -> IO ()
+-- ^ Send message to destination; the destination must not response (otherwise future 'call's will get these responses instead of their own).
+-- Throw IOError and close pipeline if send fails
+psend p@Pipeline{..} message = withMVar vStream (flip writeStream message) `onException` close p
+
+pcall :: Pipeline i o -> o -> IO (IO i)
+-- ^ Send message to destination and return /promise/ of response from one message only. The destination must reply to the message (otherwise promises will have the wrong responses in them).
+-- Throw IOError and closes pipeline if send fails, likewise for promised response.
+pcall p@Pipeline{..} message = withMVar vStream doCall `onException` close p  where
+    doCall stream = do
+        writeStream stream message
+        var <- newEmptyMVar
+        liftIO $ writeChan responseQueue var
+        return $ readMVar var >>= either throwIO return -- return promise
+
 -- * Pipe
 
 type Pipe = Pipeline Response Message
@@ -73,13 +173,13 @@ newPipeWith conn = newPipeline $ IOStream (writeMessage conn)
 
 send :: Pipe -> [Notice] -> IO ()
 -- ^ Send notices as a contiguous batch to server with no reply. Throw IOError if connection fails.
-send pipe notices = P.send pipe (notices, Nothing)
+send pipe notices = psend pipe (notices, Nothing)
 
 call :: Pipe -> [Notice] -> Request -> IO (IO Reply)
 -- ^ Send notices and request as a contiguous batch to server and return reply promise, which will block when invoked until reply arrives. This call and resulting promise will throw IOError if connection fails.
 call pipe notices request = do
     requestId <- genRequestId
-    promise <- P.call pipe (notices, Just (request, requestId))
+    promise <- pcall pipe (notices, Just (request, requestId))
     return $ check requestId <$> promise
  where
     check requestId (responseTo, reply) = if requestId == responseTo then reply else
diff --git a/System/IO/Pipeline.hs b/System/IO/Pipeline.hs
deleted file mode 100644
index e6de2a0..0000000
--- a/System/IO/Pipeline.hs
+++ /dev/null
@@ -1,118 +0,0 @@
-{- | Pipelining is sending multiple requests over a socket and receiving the responses later in the same order (a' la HTTP pipelining). This is faster than sending one request, waiting for the response, then sending the next request, and so on. This implementation returns a /promise (future)/ response for each request that when invoked waits for the response if not already arrived. Multiple threads can send on the same pipeline (and get promises back); it will send each thread's request right away without waiting.
-
-A pipeline closes itself when a read or write causes an error, so you can detect a broken pipeline by checking isClosed.  It also closes itself when garbage collected, or you can close it explicitly. -}
-
-{-# LANGUAGE RecordWildCards, NamedFieldPuns, ScopedTypeVariables #-}
-{-# LANGUAGE CPP, FlexibleContexts #-}
-
-#if (__GLASGOW_HASKELL__ >= 706)
-{-# LANGUAGE RecursiveDo #-}
-#else
-{-# LANGUAGE DoRec #-}
-#endif
-
-module System.IO.Pipeline (
-    -- * IOStream
-    IOStream(..),
-    -- * Pipeline
-    Pipeline, newPipeline, send, call, close, isClosed
-) where
-
-import Prelude hiding (length)
-import Control.Concurrent (ThreadId, forkIO, killThread)
-import Control.Concurrent.Chan (Chan, newChan, readChan, writeChan)
-import Control.Monad (forever)
-import GHC.Conc (ThreadStatus(..), threadStatus)
-
-import Control.Monad.Trans (liftIO)
-#if MIN_VERSION_base(4,6,0)
-import Control.Concurrent.MVar.Lifted (MVar, newEmptyMVar, newMVar, withMVar,
-                                       putMVar, readMVar, mkWeakMVar)
-#else
-import Control.Concurrent.MVar.Lifted (MVar, newEmptyMVar, newMVar, withMVar,
-                                         putMVar, readMVar, addMVarFinalizer)
-#endif
-import Control.Exception.Lifted (onException, throwIO, try)
-
-#if !MIN_VERSION_base(4,6,0)
-mkWeakMVar :: MVar a -> IO () -> IO ()
-mkWeakMVar = addMVarFinalizer
-#endif
-
--- * IOStream
-
--- | An IO sink and source where value of type @o@ are sent and values of type @i@ are received.
-data IOStream i o = IOStream {
-    writeStream :: o -> IO (),
-    readStream :: IO i,
-    closeStream :: IO () }
-
--- * Pipeline
-
--- | Thread-safe and pipelined connection
-data Pipeline i o = Pipeline {
-    vStream :: MVar (IOStream i o),  -- ^ Mutex on handle, so only one thread at a time can write to it
-    responseQueue :: Chan (MVar (Either IOError i)),  -- ^ Queue of threads waiting for responses. Every time a response arrive we pop the next thread and give it the response.
-    listenThread :: ThreadId
-    }
-
--- | Create new Pipeline over given handle. You should 'close' pipeline when finished, which will also close handle. If pipeline is not closed but eventually garbage collected, it will be closed along with handle.
-newPipeline :: IOStream i o -> IO (Pipeline i o)
-newPipeline stream = do
-    vStream <- newMVar stream
-    responseQueue <- newChan
-    rec
-        let pipe = Pipeline{..}
-        listenThread <- forkIO (listen pipe)
-    _ <- mkWeakMVar vStream $ do
-        killThread listenThread
-        closeStream stream
-    return pipe
-
-close :: Pipeline i o -> IO ()
--- ^ Close pipe and underlying connection
-close Pipeline{..} = do
-    killThread listenThread
-    closeStream =<< readMVar vStream
-
-isClosed :: Pipeline i o -> IO Bool
-isClosed Pipeline{listenThread} = do
-    status <- threadStatus listenThread
-    return $ case status of
-        ThreadRunning -> False
-        ThreadFinished -> True
-        ThreadBlocked _ -> False
-        ThreadDied -> True
---isPipeClosed Pipeline{..} = isClosed =<< readMVar vHandle  -- isClosed hangs while listen loop is waiting on read
-
-listen :: Pipeline i o -> IO ()
--- ^ Listen for responses and supply them to waiting threads in order
-listen Pipeline{..} = do
-    stream <- readMVar vStream
-    forever $ do
-        e <- try $ readStream stream
-        var <- readChan responseQueue
-        putMVar var e
-        case e of
-            Left err -> closeStream stream >> ioError err  -- close and stop looping
-            Right _ -> return ()
-
-send :: Pipeline i o -> o -> IO ()
--- ^ Send message to destination; the destination must not response (otherwise future 'call's will get these responses instead of their own).
--- Throw IOError and close pipeline if send fails
-send p@Pipeline{..} message = withMVar vStream (flip writeStream message) `onException` close p
-
-call :: Pipeline i o -> o -> IO (IO i)
--- ^ Send message to destination and return /promise/ of response from one message only. The destination must reply to the message (otherwise promises will have the wrong responses in them).
--- Throw IOError and closes pipeline if send fails, likewise for promised response.
-call p@Pipeline{..} message = withMVar vStream doCall `onException` close p  where
-    doCall stream = do
-        writeStream stream message
-        var <- newEmptyMVar
-        liftIO $ writeChan responseQueue var
-        return $ readMVar var >>= either throwIO return -- return promise
-
-
-{- Authors: Tony Hannan <tony@10gen.com>
-   Copyright 2011 10gen Inc.
-   Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at: http://www.apache.org/licenses/LICENSE-2.0. Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License. -}
diff --git a/mongoDB.cabal b/mongoDB.cabal
index e7cc087..a0f4989 100644
--- a/mongoDB.cabal
+++ b/mongoDB.cabal
@@ -56,7 +56,6 @@ Library
                     Database.MongoDB.Internal.Protocol
                     Database.MongoDB.Internal.Util
                     Database.MongoDB.Query
-                    System.IO.Pipeline
 
 Source-repository head
     Type:     git

From 73dfdb0b7fcb78707165074b04418282d78eff3e Mon Sep 17 00:00:00 2001
From: Victor Denisov <denisovenator@gmail.com>
Date: Sun, 10 Apr 2016 17:55:58 -0700
Subject: [PATCH 06/14] Remove IOStream from Internal.Protocol

---
 Database/MongoDB/Internal/Connection.hs |  7 ----
 Database/MongoDB/Internal/Protocol.hs   | 44 ++++++++++---------------
 2 files changed, 17 insertions(+), 34 deletions(-)

diff --git a/Database/MongoDB/Internal/Connection.hs b/Database/MongoDB/Internal/Connection.hs
index c0abdd7..a55f0a6 100644
--- a/Database/MongoDB/Internal/Connection.hs
+++ b/Database/MongoDB/Internal/Connection.hs
@@ -4,20 +4,13 @@
 
 module Database.MongoDB.Internal.Connection (
     Connection(..),
-    readExactly,
     fromHandle,
 ) where
 
 import Prelude hiding (read)
-import Data.Monoid
-import Data.IORef
 import Data.ByteString (ByteString)
 import qualified Data.ByteString as ByteString
-import qualified Data.ByteString.Lazy as Lazy (ByteString)
-import qualified Data.ByteString.Lazy as Lazy.ByteString
-import Control.Monad
 import System.IO
-import System.IO.Error (mkIOError, eofErrorType)
 
 -- | Abstract connection interface
 --
diff --git a/Database/MongoDB/Internal/Protocol.hs b/Database/MongoDB/Internal/Protocol.hs
index 8385659..fffbb88 100644
--- a/Database/MongoDB/Internal/Protocol.hs
+++ b/Database/MongoDB/Internal/Protocol.hs
@@ -83,25 +83,17 @@ mkWeakMVar :: MVar a -> IO () -> IO ()
 mkWeakMVar = addMVarFinalizer
 #endif
 
--- * IOStream
-
--- | An IO sink and source where value of type @o@ are sent and values of type @i@ are received.
-data IOStream i o = IOStream {
-    writeStream :: o -> IO (),
-    readStream :: IO i,
-    closeStream :: IO () }
-
 -- * Pipeline
 
 -- | Thread-safe and pipelined connection
-data Pipeline i o = Pipeline {
-    vStream :: MVar (IOStream i o),  -- ^ Mutex on handle, so only one thread at a time can write to it
-    responseQueue :: Chan (MVar (Either IOError i)),  -- ^ Queue of threads waiting for responses. Every time a response arrive we pop the next thread and give it the response.
+data Pipeline = Pipeline {
+    vStream :: MVar Connection,  -- ^ Mutex on handle, so only one thread at a time can write to it
+    responseQueue :: Chan (MVar (Either IOError Response)),  -- ^ Queue of threads waiting for responses. Every time a response arrive we pop the next thread and give it the response.
     listenThread :: ThreadId
     }
 
 -- | Create new Pipeline over given handle. You should 'close' pipeline when finished, which will also close handle. If pipeline is not closed but eventually garbage collected, it will be closed along with handle.
-newPipeline :: IOStream i o -> IO (Pipeline i o)
+newPipeline :: Connection -> IO Pipeline
 newPipeline stream = do
     vStream <- newMVar stream
     responseQueue <- newChan
@@ -110,16 +102,16 @@ newPipeline stream = do
         listenThread <- forkIO (listen pipe)
     _ <- mkWeakMVar vStream $ do
         killThread listenThread
-        closeStream stream
+        Connection.close stream
     return pipe
 
-close :: Pipeline i o -> IO ()
+close :: Pipeline -> IO ()
 -- ^ Close pipe and underlying connection
 close Pipeline{..} = do
     killThread listenThread
-    closeStream =<< readMVar vStream
+    Connection.close =<< readMVar vStream
 
-isClosed :: Pipeline i o -> IO Bool
+isClosed :: Pipeline -> IO Bool
 isClosed Pipeline{listenThread} = do
     status <- threadStatus listenThread
     return $ case status of
@@ -129,36 +121,36 @@ isClosed Pipeline{listenThread} = do
         ThreadDied -> True
 --isPipeClosed Pipeline{..} = isClosed =<< readMVar vHandle  -- isClosed hangs while listen loop is waiting on read
 
-listen :: Pipeline i o -> IO ()
+listen :: Pipeline -> IO ()
 -- ^ Listen for responses and supply them to waiting threads in order
 listen Pipeline{..} = do
     stream <- readMVar vStream
     forever $ do
-        e <- try $ readStream stream
+        e <- try $ readMessage stream
         var <- readChan responseQueue
         putMVar var e
         case e of
-            Left err -> closeStream stream >> ioError err  -- close and stop looping
+            Left err -> Connection.close stream >> ioError err  -- close and stop looping
             Right _ -> return ()
 
-psend :: Pipeline i o -> o -> IO ()
+psend :: Pipeline -> Message -> IO ()
 -- ^ Send message to destination; the destination must not response (otherwise future 'call's will get these responses instead of their own).
 -- Throw IOError and close pipeline if send fails
-psend p@Pipeline{..} message = withMVar vStream (flip writeStream message) `onException` close p
+psend p@Pipeline{..} message = withMVar vStream (flip writeMessage message) `onException` close p
 
-pcall :: Pipeline i o -> o -> IO (IO i)
+pcall :: Pipeline -> Message -> IO (IO Response)
 -- ^ Send message to destination and return /promise/ of response from one message only. The destination must reply to the message (otherwise promises will have the wrong responses in them).
 -- Throw IOError and closes pipeline if send fails, likewise for promised response.
 pcall p@Pipeline{..} message = withMVar vStream doCall `onException` close p  where
     doCall stream = do
-        writeStream stream message
+        writeMessage stream message
         var <- newEmptyMVar
         liftIO $ writeChan responseQueue var
         return $ readMVar var >>= either throwIO return -- return promise
 
 -- * Pipe
 
-type Pipe = Pipeline Response Message
+type Pipe = Pipeline
 -- ^ Thread-safe TCP connection with pipelined requests
 
 newPipe :: Handle -> IO Pipe
@@ -167,9 +159,7 @@ newPipe handle = Connection.fromHandle handle >>= newPipeWith
 
 newPipeWith :: Connection -> IO Pipe
 -- ^ Create pipe over connection
-newPipeWith conn = newPipeline $ IOStream (writeMessage conn)
-                                          (readMessage conn)
-                                          (Connection.close conn)
+newPipeWith conn = newPipeline conn
 
 send :: Pipe -> [Notice] -> IO ()
 -- ^ Send notices as a contiguous batch to server with no reply. Throw IOError if connection fails.

From 655f4b90bc8a5687154a8b0e2518993ef09c4cf4 Mon Sep 17 00:00:00 2001
From: Victor Denisov <denisovenator@gmail.com>
Date: Sun, 1 May 2016 19:05:51 -0700
Subject: [PATCH 07/14] Rename readExactly to read

---
 Database/MongoDB/Internal/Connection.hs | 4 ++--
 Database/MongoDB/Internal/Protocol.hs   | 4 ++--
 Database/MongoDB/Internal/Tls.hs        | 2 +-
 3 files changed, 5 insertions(+), 5 deletions(-)

diff --git a/Database/MongoDB/Internal/Connection.hs b/Database/MongoDB/Internal/Connection.hs
index a55f0a6..8679c5e 100644
--- a/Database/MongoDB/Internal/Connection.hs
+++ b/Database/MongoDB/Internal/Connection.hs
@@ -16,7 +16,7 @@ import System.IO
 --
 -- `read` should return `ByteString.null` on EOF
 data Connection = Connection {
-    readExactly :: Int -> IO ByteString,
+    read  :: Int -> IO ByteString,
     write :: ByteString -> IO (),
     flush :: IO (),
     close :: IO ()}
@@ -25,7 +25,7 @@ fromHandle :: Handle -> IO Connection
 -- ^ Make connection form handle
 fromHandle handle = do
   return Connection
-    { readExactly = ByteString.hGet handle
+    { read  = ByteString.hGet handle
     , write = ByteString.hPut handle
     , flush = hFlush handle
     , close = hClose handle
diff --git a/Database/MongoDB/Internal/Protocol.hs b/Database/MongoDB/Internal/Protocol.hs
index fffbb88..4c660d8 100644
--- a/Database/MongoDB/Internal/Protocol.hs
+++ b/Database/MongoDB/Internal/Protocol.hs
@@ -207,8 +207,8 @@ readMessage :: Connection -> IO Response
 -- ^ read response from a connection
 readMessage conn = readResp  where
     readResp = do
-        len <- fromEnum . decodeSize . L.fromStrict <$> Connection.readExactly conn 4
-        runGet getReply . L.fromStrict <$> Connection.readExactly conn len
+        len <- fromEnum . decodeSize . L.fromStrict <$> Connection.read conn 4
+        runGet getReply . L.fromStrict <$> Connection.read conn len
     decodeSize = subtract 4 . runGet getInt32
 
 type FullCollection = Text
diff --git a/Database/MongoDB/Internal/Tls.hs b/Database/MongoDB/Internal/Tls.hs
index 3579a9b..2b7ae26 100644
--- a/Database/MongoDB/Internal/Tls.hs
+++ b/Database/MongoDB/Internal/Tls.hs
@@ -56,7 +56,7 @@ tlsConnection :: TLS.Context -> IO () -> IO Connection
 tlsConnection ctx close = do
   restRef <- newIORef mempty
   return Connection
-    { Connection.readExactly = \count -> let
+    { Connection.read = \count -> let
           readSome = do
             rest <- readIORef restRef
             writeIORef restRef mempty

From a4e5726e1e3f4ac1a886c67c583e9365f26919ad Mon Sep 17 00:00:00 2001
From: Victor Denisov <denisovenator@gmail.com>
Date: Sun, 1 May 2016 19:11:02 -0700
Subject: [PATCH 08/14] Clean up redundant warnings

---
 Database/MongoDB/Internal/Protocol.hs | 5 +----
 1 file changed, 1 insertion(+), 4 deletions(-)

diff --git a/Database/MongoDB/Internal/Protocol.hs b/Database/MongoDB/Internal/Protocol.hs
index 4c660d8..0492794 100644
--- a/Database/MongoDB/Internal/Protocol.hs
+++ b/Database/MongoDB/Internal/Protocol.hs
@@ -34,7 +34,6 @@ module Database.MongoDB.Internal.Protocol (
 #if !MIN_VERSION_base(4,8,0)
 import Control.Applicative ((<$>))
 #endif
-import Control.Arrow ((***))
 import Control.Monad (forM, replicateM, unless)
 import Data.Binary.Get (Get, runGet)
 import Data.Binary.Put (Put, runPut)
@@ -50,7 +49,6 @@ import Control.Concurrent.Chan (Chan, newChan, readChan, writeChan)
 import Control.Concurrent (ThreadId, forkIO, killThread)
 
 import Control.Exception.Lifted (onException, throwIO, try)
-import qualified Control.Exception.Lifted as CEL
 
 import qualified Data.ByteString.Lazy as L
 
@@ -64,8 +62,7 @@ import qualified Crypto.Hash.MD5 as MD5
 import qualified Data.Text as T
 import qualified Data.Text.Encoding as TE
 
-import Database.MongoDB.Internal.Util (whenJust, bitOr, byteStringHex)
-import System.IO (hClose, hFlush)
+import Database.MongoDB.Internal.Util (bitOr, byteStringHex)
 
 import Database.MongoDB.Internal.Connection (Connection)
 import qualified Database.MongoDB.Internal.Connection as Connection

From ccd6727ab4cd49f559bf28e94ec7343afe7e2fca Mon Sep 17 00:00:00 2001
From: Victor Denisov <denisovenator@gmail.com>
Date: Sun, 1 May 2016 19:23:30 -0700
Subject: [PATCH 09/14] Use conventional types for Tls.Connect

---
 Database/MongoDB/Internal/Tls.hs | 11 ++++-------
 1 file changed, 4 insertions(+), 7 deletions(-)

diff --git a/Database/MongoDB/Internal/Tls.hs b/Database/MongoDB/Internal/Tls.hs
index 2b7ae26..f78ed65 100644
--- a/Database/MongoDB/Internal/Tls.hs
+++ b/Database/MongoDB/Internal/Tls.hs
@@ -11,8 +11,6 @@ where
 
 import Data.IORef
 import Data.Monoid
-import Data.Text (Text)
-import qualified Data.Text as Text
 import qualified Data.ByteString as ByteString
 import qualified Data.ByteString.Lazy as Lazy.ByteString
 import Data.Default.Class (def)
@@ -25,20 +23,19 @@ import Database.MongoDB.Internal.Protocol (newPipeWith)
 import Database.MongoDB.Internal.Connection (Connection(Connection))
 import qualified Database.MongoDB.Internal.Connection as Connection
 import System.IO.Error (mkIOError, eofErrorType)
-import qualified Network
+import Network (connectTo, HostName, PortID)
 import qualified Network.TLS as TLS
 import qualified Network.TLS.Extra.Cipher as TLS
 import qualified Control.IO.Region as Region
 
 -- | Connect to mongodb using TLS
-connect :: Text -> Int -> IO Pipe
+connect :: HostName -> PortID -> IO Pipe
 connect host port = bracketOnError Region.open Region.close $ \r -> do
   handle <- Region.alloc_ r
-    (Network.connectTo (Text.unpack host)
-                       (Network.PortNumber $ fromIntegral port))
+    (connectTo host port)
     hClose
 
-  let params = (TLS.defaultParamsClient (Text.unpack host) "")
+  let params = (TLS.defaultParamsClient host "")
         { TLS.clientSupported = def
             { TLS.supportedCiphers = TLS.ciphersuite_all}
         , TLS.clientHooks = def

From 89ee88e67c8faa383d8b831abbeed8d2cf72a43f Mon Sep 17 00:00:00 2001
From: Victor Denisov <denisovenator@gmail.com>
Date: Mon, 2 May 2016 21:30:00 -0700
Subject: [PATCH 10/14] Rename Internal.Connection module to Transport

---
 Database/MongoDB/Internal/Protocol.hs         | 30 +++++++++----------
 Database/MongoDB/Internal/Tls.hs              | 18 +++++------
 .../{Internal/Connection.hs => Transport.hs}  | 12 ++++----
 mongoDB.cabal                                 |  2 +-
 4 files changed, 31 insertions(+), 31 deletions(-)
 rename Database/MongoDB/{Internal/Connection.hs => Transport.hs} (77%)

diff --git a/Database/MongoDB/Internal/Protocol.hs b/Database/MongoDB/Internal/Protocol.hs
index 0492794..06b54f6 100644
--- a/Database/MongoDB/Internal/Protocol.hs
+++ b/Database/MongoDB/Internal/Protocol.hs
@@ -64,8 +64,8 @@ import qualified Data.Text.Encoding as TE
 
 import Database.MongoDB.Internal.Util (bitOr, byteStringHex)
 
-import Database.MongoDB.Internal.Connection (Connection)
-import qualified Database.MongoDB.Internal.Connection as Connection
+import Database.MongoDB.Transport (Transport)
+import qualified Database.MongoDB.Transport as T
 
 #if MIN_VERSION_base(4,6,0)
 import Control.Concurrent.MVar.Lifted (MVar, newEmptyMVar, newMVar, withMVar,
@@ -84,13 +84,13 @@ mkWeakMVar = addMVarFinalizer
 
 -- | Thread-safe and pipelined connection
 data Pipeline = Pipeline {
-    vStream :: MVar Connection,  -- ^ Mutex on handle, so only one thread at a time can write to it
+    vStream :: MVar Transport,  -- ^ Mutex on handle, so only one thread at a time can write to it
     responseQueue :: Chan (MVar (Either IOError Response)),  -- ^ Queue of threads waiting for responses. Every time a response arrive we pop the next thread and give it the response.
     listenThread :: ThreadId
     }
 
 -- | Create new Pipeline over given handle. You should 'close' pipeline when finished, which will also close handle. If pipeline is not closed but eventually garbage collected, it will be closed along with handle.
-newPipeline :: Connection -> IO Pipeline
+newPipeline :: Transport -> IO Pipeline
 newPipeline stream = do
     vStream <- newMVar stream
     responseQueue <- newChan
@@ -99,14 +99,14 @@ newPipeline stream = do
         listenThread <- forkIO (listen pipe)
     _ <- mkWeakMVar vStream $ do
         killThread listenThread
-        Connection.close stream
+        T.close stream
     return pipe
 
 close :: Pipeline -> IO ()
 -- ^ Close pipe and underlying connection
 close Pipeline{..} = do
     killThread listenThread
-    Connection.close =<< readMVar vStream
+    T.close =<< readMVar vStream
 
 isClosed :: Pipeline -> IO Bool
 isClosed Pipeline{listenThread} = do
@@ -127,7 +127,7 @@ listen Pipeline{..} = do
         var <- readChan responseQueue
         putMVar var e
         case e of
-            Left err -> Connection.close stream >> ioError err  -- close and stop looping
+            Left err -> T.close stream >> ioError err  -- close and stop looping
             Right _ -> return ()
 
 psend :: Pipeline -> Message -> IO ()
@@ -152,9 +152,9 @@ type Pipe = Pipeline
 
 newPipe :: Handle -> IO Pipe
 -- ^ Create pipe over handle
-newPipe handle = Connection.fromHandle handle >>= newPipeWith
+newPipe handle = T.fromHandle handle >>= newPipeWith
 
-newPipeWith :: Connection -> IO Pipe
+newPipeWith :: Transport -> IO Pipe
 -- ^ Create pipe over connection
 newPipeWith conn = newPipeline conn
 
@@ -178,7 +178,7 @@ type Message = ([Notice], Maybe (Request, RequestId))
 -- ^ A write notice(s) with getLastError request, or just query request.
 -- Note, that requestId will be out of order because request ids will be generated for notices after the request id supplied was generated. This is ok because the mongo server does not care about order just uniqueness.
 
-writeMessage :: Connection -> Message -> IO ()
+writeMessage :: Transport -> Message -> IO ()
 -- ^ Write message to connection
 writeMessage conn (notices, mRequest) = do
     noticeStrings <- forM notices $ \n -> do
@@ -191,8 +191,8 @@ writeMessage conn (notices, mRequest) = do
           let s = runPut $ putRequest request requestId
           return $ (lenBytes s) `L.append` s
 
-    Connection.write conn $ L.toStrict $ L.concat $ noticeStrings ++ (maybeToList requestString)
-    Connection.flush conn
+    T.write conn $ L.toStrict $ L.concat $ noticeStrings ++ (maybeToList requestString)
+    T.flush conn
  where
     lenBytes bytes = encodeSize . toEnum . fromEnum $ L.length bytes
     encodeSize = runPut . putInt32 . (+ 4)
@@ -200,12 +200,12 @@ writeMessage conn (notices, mRequest) = do
 type Response = (ResponseTo, Reply)
 -- ^ Message received from a Mongo server in response to a Request
 
-readMessage :: Connection -> IO Response
+readMessage :: Transport -> IO Response
 -- ^ read response from a connection
 readMessage conn = readResp  where
     readResp = do
-        len <- fromEnum . decodeSize . L.fromStrict <$> Connection.read conn 4
-        runGet getReply . L.fromStrict <$> Connection.read conn len
+        len <- fromEnum . decodeSize . L.fromStrict <$> T.read conn 4
+        runGet getReply . L.fromStrict <$> T.read conn len
     decodeSize = subtract 4 . runGet getInt32
 
 type FullCollection = Text
diff --git a/Database/MongoDB/Internal/Tls.hs b/Database/MongoDB/Internal/Tls.hs
index f78ed65..2f51540 100644
--- a/Database/MongoDB/Internal/Tls.hs
+++ b/Database/MongoDB/Internal/Tls.hs
@@ -20,8 +20,8 @@ import Control.Monad (when, unless)
 import System.IO
 import Database.MongoDB (Pipe)
 import Database.MongoDB.Internal.Protocol (newPipeWith)
-import Database.MongoDB.Internal.Connection (Connection(Connection))
-import qualified Database.MongoDB.Internal.Connection as Connection
+import Database.MongoDB.Transport (Transport(Transport))
+import qualified Database.MongoDB.Transport as T
 import System.IO.Error (mkIOError, eofErrorType)
 import Network (connectTo, HostName, PortID)
 import qualified Network.TLS as TLS
@@ -49,11 +49,11 @@ connect host port = bracketOnError Region.open Region.close $ \r -> do
   conn <- tlsConnection context (Region.close r)
   newPipeWith conn
 
-tlsConnection :: TLS.Context -> IO () -> IO Connection
+tlsConnection :: TLS.Context -> IO () -> IO Transport
 tlsConnection ctx close = do
   restRef <- newIORef mempty
-  return Connection
-    { Connection.read = \count -> let
+  return Transport
+    { T.read = \count -> let
           readSome = do
             rest <- readIORef restRef
             writeIORef restRef mempty
@@ -75,10 +75,10 @@ tlsConnection ctx close = do
                   unread rest
                 return (acc <> Lazy.ByteString.fromStrict res)
               else go (acc <> Lazy.ByteString.fromStrict chunk) (n - len)
-          eof = mkIOError eofErrorType "Database.MongoDB.Internal.Connection"
+          eof = mkIOError eofErrorType "Database.MongoDB.Transport"
                 Nothing Nothing
        in Lazy.ByteString.toStrict <$> go mempty count
-    , Connection.write = TLS.sendData ctx . Lazy.ByteString.fromStrict
-    , Connection.flush = TLS.contextFlush ctx
-    , Connection.close = close
+    , T.write = TLS.sendData ctx . Lazy.ByteString.fromStrict
+    , T.flush = TLS.contextFlush ctx
+    , T.close = close
     }
diff --git a/Database/MongoDB/Internal/Connection.hs b/Database/MongoDB/Transport.hs
similarity index 77%
rename from Database/MongoDB/Internal/Connection.hs
rename to Database/MongoDB/Transport.hs
index 8679c5e..dffdd35 100644
--- a/Database/MongoDB/Internal/Connection.hs
+++ b/Database/MongoDB/Transport.hs
@@ -2,8 +2,8 @@
 -- | This module defines a connection interface. It could be a regular
 -- network connection, TLS connection, a mock or anything else.
 
-module Database.MongoDB.Internal.Connection (
-    Connection(..),
+module Database.MongoDB.Transport (
+    Transport(..),
     fromHandle,
 ) where
 
@@ -12,19 +12,19 @@ import Data.ByteString (ByteString)
 import qualified Data.ByteString as ByteString
 import System.IO
 
--- | Abstract connection interface
+-- | Abstract transport interface
 --
 -- `read` should return `ByteString.null` on EOF
-data Connection = Connection {
+data Transport = Transport {
     read  :: Int -> IO ByteString,
     write :: ByteString -> IO (),
     flush :: IO (),
     close :: IO ()}
 
-fromHandle :: Handle -> IO Connection
+fromHandle :: Handle -> IO Transport
 -- ^ Make connection form handle
 fromHandle handle = do
-  return Connection
+  return Transport
     { read  = ByteString.hGet handle
     , write = ByteString.hPut handle
     , flush = hFlush handle
diff --git a/mongoDB.cabal b/mongoDB.cabal
index a0f4989..3df9215 100644
--- a/mongoDB.cabal
+++ b/mongoDB.cabal
@@ -51,11 +51,11 @@ Library
   Exposed-modules:  Database.MongoDB
                     Database.MongoDB.Admin
                     Database.MongoDB.Connection
-                    Database.MongoDB.Internal.Connection
                     Database.MongoDB.Internal.Tls
                     Database.MongoDB.Internal.Protocol
                     Database.MongoDB.Internal.Util
                     Database.MongoDB.Query
+                    Database.MongoDB.Transport
 
 Source-repository head
     Type:     git

From e38ec59d1eaa935acd4bb50ffd8d6ab6a07867ad Mon Sep 17 00:00:00 2001
From: Victor Denisov <denisovenator@gmail.com>
Date: Mon, 2 May 2016 22:05:02 -0700
Subject: [PATCH 11/14] Move Tls module to Transport

---
 Database/MongoDB/{Internal => Transport}/Tls.hs | 2 +-
 mongoDB.cabal                                   | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)
 rename Database/MongoDB/{Internal => Transport}/Tls.hs (98%)

diff --git a/Database/MongoDB/Internal/Tls.hs b/Database/MongoDB/Transport/Tls.hs
similarity index 98%
rename from Database/MongoDB/Internal/Tls.hs
rename to Database/MongoDB/Transport/Tls.hs
index 2f51540..9029d2a 100644
--- a/Database/MongoDB/Internal/Tls.hs
+++ b/Database/MongoDB/Transport/Tls.hs
@@ -3,7 +3,7 @@
 
 -- | TLS connection to mongodb
 
-module Database.MongoDB.Internal.Tls
+module Database.MongoDB.Transport.Tls
 (
   connect,
 )
diff --git a/mongoDB.cabal b/mongoDB.cabal
index 3df9215..2f2de39 100644
--- a/mongoDB.cabal
+++ b/mongoDB.cabal
@@ -51,11 +51,11 @@ Library
   Exposed-modules:  Database.MongoDB
                     Database.MongoDB.Admin
                     Database.MongoDB.Connection
-                    Database.MongoDB.Internal.Tls
                     Database.MongoDB.Internal.Protocol
                     Database.MongoDB.Internal.Util
                     Database.MongoDB.Query
                     Database.MongoDB.Transport
+                    Database.MongoDB.Transport.Tls
 
 Source-repository head
     Type:     git

From f18e4ff9f8b59d1d09442931ad42c26e486740b5 Mon Sep 17 00:00:00 2001
From: Victor Denisov <denisovenator@gmail.com>
Date: Mon, 2 May 2016 23:14:18 -0700
Subject: [PATCH 12/14] Update CHANGELOG

---
 CHANGELOG.md | 8 ++++++++
 1 file changed, 8 insertions(+)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 4cd4982..c745912 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,14 @@
 All notable changes to this project will be documented in this file.
 This project adheres to [Package Versioning Policy](https://wiki.haskell.org/Package_versioning_policy).
 
+## [2.1.0] - unreleased
+
+### Added
+- TLS implementation. So far it is an experimental feature.
+
+### Removed
+- System.IO.Pipeline module
+
 ## [2.0.10] - 2015-12-22
 
 ### Fixed

From 5ca93be5e8e1bf71e90971b1f5d53eafd6dd59f5 Mon Sep 17 00:00:00 2001
From: Victor Denisov <denisovenator@gmail.com>
Date: Tue, 3 May 2016 23:02:54 -0700
Subject: [PATCH 13/14] Drop io-region dependency

---
 Database/MongoDB/Transport/Tls.hs | 34 ++++++++++++++++---------------
 mongoDB.cabal                     |  1 -
 2 files changed, 18 insertions(+), 17 deletions(-)

diff --git a/Database/MongoDB/Transport/Tls.hs b/Database/MongoDB/Transport/Tls.hs
index 9029d2a..0e0b4ec 100644
--- a/Database/MongoDB/Transport/Tls.hs
+++ b/Database/MongoDB/Transport/Tls.hs
@@ -1,12 +1,20 @@
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE RecordWildCards #-}
 
--- | TLS connection to mongodb
+{-|
+Module      : MongoDB TLS
+Description : TLS transport to mongodb
+Copyright   : (c)	Yuras Shumovich, 2016
+License     : Apache 2.0
+Maintainer  : Victor Denisov denisovenator@gmail.com
+Stability   : experimental
+Portability : POSIX
 
+This module is for connecting to TLS enabled mongodb servers.
+Be aware that this module is highly experimental and is barely tested.
+-}
 module Database.MongoDB.Transport.Tls
-(
-  connect,
-)
+(connect)
 where
 
 import Data.IORef
@@ -26,14 +34,10 @@ import System.IO.Error (mkIOError, eofErrorType)
 import Network (connectTo, HostName, PortID)
 import qualified Network.TLS as TLS
 import qualified Network.TLS.Extra.Cipher as TLS
-import qualified Control.IO.Region as Region
 
 -- | Connect to mongodb using TLS
 connect :: HostName -> PortID -> IO Pipe
-connect host port = bracketOnError Region.open Region.close $ \r -> do
-  handle <- Region.alloc_ r
-    (connectTo host port)
-    hClose
+connect host port = bracketOnError (connectTo host port) hClose $ \handle -> do
 
   let params = (TLS.defaultParamsClient host "")
         { TLS.clientSupported = def
@@ -41,16 +45,14 @@ connect host port = bracketOnError Region.open Region.close $ \r -> do
         , TLS.clientHooks = def
             { TLS.onServerCertificate = \_ _ _ _ -> return []}
         }
-  context <- Region.alloc_ r
-    (TLS.contextNew handle params)
-    TLS.contextClose
+  context <- TLS.contextNew handle params
   TLS.handshake context
 
-  conn <- tlsConnection context (Region.close r)
+  conn <- tlsConnection context
   newPipeWith conn
 
-tlsConnection :: TLS.Context -> IO () -> IO Transport
-tlsConnection ctx close = do
+tlsConnection :: TLS.Context -> IO Transport
+tlsConnection ctx = do
   restRef <- newIORef mempty
   return Transport
     { T.read = \count -> let
@@ -80,5 +82,5 @@ tlsConnection ctx close = do
        in Lazy.ByteString.toStrict <$> go mempty count
     , T.write = TLS.sendData ctx . Lazy.ByteString.fromStrict
     , T.flush = TLS.contextFlush ctx
-    , T.close = close
+    , T.close = TLS.contextClose ctx
     }
diff --git a/mongoDB.cabal b/mongoDB.cabal
index 2f2de39..b389aef 100644
--- a/mongoDB.cabal
+++ b/mongoDB.cabal
@@ -34,7 +34,6 @@ Library
                     , mtl >= 2
                     , cryptohash -any
                     , network -any
-                    , io-region -any
                     , parsec -any
                     , random -any
                     , random-shuffle -any

From 8132604443fc507a82869afe23807628ad51920e Mon Sep 17 00:00:00 2001
From: Victor Denisov <denisovenator@gmail.com>
Date: Tue, 3 May 2016 23:43:55 -0700
Subject: [PATCH 14/14] Add warning about experimental TLS support

---
 Database/MongoDB/Transport/Tls.hs | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git a/Database/MongoDB/Transport/Tls.hs b/Database/MongoDB/Transport/Tls.hs
index 0e0b4ec..2fd3c73 100644
--- a/Database/MongoDB/Transport/Tls.hs
+++ b/Database/MongoDB/Transport/Tls.hs
@@ -3,7 +3,7 @@
 
 {-|
 Module      : MongoDB TLS
-Description : TLS transport to mongodb
+Description : TLS transport for mongodb
 Copyright   : (c)	Yuras Shumovich, 2016
 License     : Apache 2.0
 Maintainer  : Victor Denisov denisovenator@gmail.com
@@ -11,7 +11,9 @@ Stability   : experimental
 Portability : POSIX
 
 This module is for connecting to TLS enabled mongodb servers.
-Be aware that this module is highly experimental and is barely tested.
+ATTENTION!!! Be aware that this module is highly experimental and is
+barely tested. The current implementation doesn't verify server's identity.
+It only allows you to connect to a mongodb server using TLS protocol.
 -}
 module Database.MongoDB.Transport.Tls
 (connect)