initial commit

1 files changed, 859 insertions, 0 deletions

diff --git a/src/Ssb/Peer/RPC.hs b/src/Ssb/Peer/RPC.hs
new file mode 100644
index 0000000..e5a4c9a
--- /dev/null
+++ b/src/Ssb/Peer/RPC.hs
@@ -0,0 +1,859 @@
+-- For more information kindly refer the to protocol guide
+-- https://ssbc.github.io/scuttlebutt-protocol-guide
+
+module Ssb.Peer.RPC where
+
+import           Protolude
+
+import           Control.Concurrent.STM
+import           Control.Monad.Fail
+import           Data.Aeson                     ( FromJSON
+                                                , ToJSON
+                                                )
+import           Data.Aeson                    as Aeson
+import qualified Data.ByteString               as BS
+import qualified Data.ByteString.Lazy          as BS
+                                                ( toStrict )
+import qualified Data.Map.Strict               as Map
+import           Data.Default
+import           Data.Either.Combinators        ( mapLeft
+                                                , mapRight
+                                                )
+import           Data.Text                     as Text
+import           Data.Serialize                as Serialize
+
+import           Ssb.Aux
+import qualified Ssb.Identity                  as Ssb
+import qualified Ssb.Peer.BoxStream            as BoxStream
+
+data BodyType = Binary | UTF8String | JSON | UnknownBodyType
+    deriving (Bounded,Eq,Show)
+
+instance Convertible Word8 BodyType where
+  convert v = case v .&. 3 of
+    0 -> Binary
+    1 -> UTF8String
+    2 -> JSON
+    v -> UnknownBodyType
+
+instance Convertible BodyType Word8 where
+  convert Binary          = 0
+  convert UTF8String      = 1
+  convert JSON            = 2
+  convert UnknownBodyType = 3
+
+data Flags = Flags
+    { unused1      :: Bool
+    , unused2      :: Bool
+    , unused3      :: Bool
+    , unused4      :: Bool
+    , isStream     :: Bool
+    , isEndOrError :: Bool
+    , bodyType     :: BodyType
+    } deriving (Eq,Show)
+
+instance Convertible Flags Word8 where
+  convert arg =
+    set (unused1 arg) 7
+      $ set (unused2 arg)      6
+      $ set (unused3 arg)      5
+      $ set (unused4 arg)      4
+      $ set (isStream arg)     3
+      $ set (isEndOrError arg) 2
+      $ convert (bodyType arg)
+   where
+    set True  pos arg = setBit arg pos
+    set False _   arg = arg
+
+instance Convertible Word8 Flags where
+  convert w = Flags { unused1      = testBit w 7
+                    , unused2      = testBit w 6
+                    , unused3      = testBit w 5
+                    , unused4      = testBit w 4
+                    , isStream     = testBit w 3
+                    , isEndOrError = testBit w 2
+                    , bodyType     = convert w
+                    }
+
+instance Default Flags where
+  def = convert (zeroBits :: Word8)
+
+instance Serialize.Serialize Flags where
+  get = convert <$> getWord8
+  put = putWord8 . convert
+
+-- | ProcedureType defines the type of remote call.
+data ProcedureType =  Async | Source | Duplex
+    deriving (Eq,Generic,Ord,Show)
+
+instance FromJSON ProcedureType where
+  parseJSON = withText "ProcedureType" $ \v -> case v of
+    "async"   -> return Async
+    "source"  -> return Source
+    "duplex"  -> return Duplex
+    otherwise -> fail $ "unknown value '" <> toS v <> "'"
+
+instance ToJSON ProcedureType where
+  toJSON Async  = "async"
+  toJSON Source = "source"
+  toJSON Duplex = "duplex"
+
+-- | HeaderLength is the length of the RPC header in bytes
+headerLength :: Int
+headerLength = 9
+
+-- | bodySizeLength is the length of the bodySize parameter in the RPC header.
+bodySizeLength :: Int
+bodySizeLength = 4
+
+-- | requestNumberLength is the length of the requestNumberLength parameter in
+-- the RPC header.
+requestNumberLength :: Int
+requestNumberLength = 4
+
+-- | Header is the first part of a RPC message used for stream control and communication.
+data Header = Header
+    { flags         :: Flags
+    , bodyLength    :: Word32
+    , requestNumber :: Int32
+    } deriving (Eq,Generic,Show)
+
+instance Serialize.Serialize Header
+
+-- | GoodByeHeader is the RPC message header signalling the end of the RPC
+-- stream.
+goodByeHeader :: Header
+goodByeHeader = Header (convert (zeroBits :: Word8)) 0 0
+
+newHeader :: Flags -> Int32 -> ByteString -> Either Text Header
+newHeader flags reqNum body = do
+  let bodyLength = fromIntegral $ BS.length body
+  return Header { flags         = flags
+                , bodyLength    = bodyLength
+                , requestNumber = reqNum
+                }
+
+-- | MessagePayload describes and contains the contents of an RPC message.
+data MessagePayload = BinaryPayload ByteString | TextPayload ByteString | JSONPayload ByteString
+    deriving (Generic,Show)
+
+-- | Message is a single message in the RPC stream
+data Message = Message
+    { header :: Header
+    , body   :: MessagePayload
+    } deriving (Generic,Show)
+
+instance Serialize.Serialize Message where
+  get = do
+    header <- Serialize.get
+    buf    <- Serialize.getByteString (fromIntegral $ bodyLength header)
+    let typ = bodyType . flags $ header
+    payload <- case typ of
+      Binary          -> return $ BinaryPayload buf
+      UTF8String      -> return $ TextPayload buf
+      JSON            -> return $ JSONPayload buf
+      UnknownBodyType -> fail "unknown body type"
+    return $ Message header payload
+
+  put msg = do
+    Serialize.put (header msg)
+    case body msg of
+      BinaryPayload buf -> Serialize.putByteString buf
+      JSONPayload   buf -> Serialize.putByteString buf
+      TextPayload   buf -> Serialize.putByteString buf
+
+-- | newJSONMessage is a convenience function to create a RPC message with a
+-- JSON payload.
+newJSONMessage :: ToJSON a => Int32 -> a -> Either Text Message
+newJSONMessage reqNum payload = do
+  let buf = BS.toStrict $ Aeson.encode payload
+  len     <- maybeWord8 $ BS.length buf
+  header' <- newHeader (def { bodyType = JSON }) reqNum buf
+  return $ Message { header = header', body = JSONPayload buf }
+
+-- | newJSONMessage is a convenience function to decode a RPC JSON message.
+decodeJSONMessage :: FromJSON a => Message -> Either Text a
+decodeJSONMessage msg = case body msg of
+  JSONPayload buf -> decodeJSON buf
+  _               -> error "unexpected body type"
+
+isRequest :: Int32 -> Message -> Bool
+isRequest nextReqNum msg = do
+  let reqNum = requestNumber . header $ msg
+  nextReqNum == reqNum
+
+-- | Request is the message representing a RPC call.
+data Request a = Request
+    { name :: [Text]
+    , typ  :: ProcedureType
+    , args :: a
+    } deriving (Show)
+
+instance (FromJSON a) => FromJSON (Request a) where
+  parseJSON = withObject "Request"
+    $ \v -> Request <$> v .: "name" <*> v .:? "type" .!= Async <*> v .: "args"
+
+instance (ToJSON a) => ToJSON (Request a) where
+  toJSON arg =
+    object ["name" .= name arg, "type" .= typ arg, "args" .= args arg]
+
+data Direction = Incoming | Outgoing
+    deriving (Eq,Generic,Show)
+
+data StreamStatus =
+      -- | Open the stream is active.
+      Open
+      -- | The (Async) stream is awaiting a response.
+    | AwaitingResponse
+      -- | The stream is closed and waiting for its peer to do the same.
+    | AwaitingCloseRecv
+      -- | The stream is closed.
+    | Closed
+    deriving (Eq,Show)
+
+data Stream  = Stream {
+      streamID    :: Int32
+    , streamType  :: ProcedureType
+    , conn        :: ConnState
+    , direction   :: Direction
+    , status      :: StreamStatus
+    , peer        :: Ssb.PublicKey
+    }
+
+formatStream :: Stream -> Text
+formatStream stream =
+  (show $ streamID stream :: Text)
+    <> "\t"
+    <> show (streamType stream)
+    <> "\t"
+    <> show (direction stream)
+    <> "\t"
+    <> show (status stream)
+
+foldStream
+  :: (a -> MessagePayload -> IO (Either Text a))
+  -> a
+  -> Stream
+  -> IO (Either Text a)
+foldStream fn acc stream = do
+  msg <- readStream stream
+  case msg of
+    (Just msg') -> do
+      acc' <- fn acc msg'
+      either (return . error) (\v -> foldStream fn v stream) acc'
+    Nothing -> return . return $ acc
+
+foldJSONStream
+  :: (FromJSON b)
+  => (a -> b -> IO (Either Text a))
+  -> a
+  -> Stream
+  -> IO (Either Text a)
+foldJSONStream fn acc stream = do
+  msg <- readStreamJSON stream
+  case msg of
+    (Right (Just msg')) -> do
+      acc' <- fn acc msg'
+      either (return . error) (\v -> foldJSONStream fn v stream) acc'
+    (Right Nothing) -> return . return $ acc
+    (Left  err    ) -> return $ error err
+
+data ConnState = ConnState {
+    connPeer           :: Ssb.PublicKey
+  , boxConn            :: BoxStream.Conn
+  , streamsIn          :: TMVar (Map Int32 (Stream, TChan (Maybe Message)))
+  , streamsOut         :: TMVar (Map Int32 (Stream, TChan (Maybe Message)))
+  , nextIncomingReqNum :: TMVar Int32
+  , nextOutgoingReqNum :: TMVar Int32
+  , lock               :: TMVar Bool
+  }
+
+-- | Endpoint identifies which Remote Procedure Call to make.
+data Endpoint = Endpoint [Text] ProcedureType
+  deriving (Eq, Ord)
+
+formatEndpoint :: Endpoint -> Text
+formatEndpoint (Endpoint paths typ) =
+  Text.intercalate "." paths <> ":" <> show typ
+
+-- | HandlerFunc's are used to serve Remove Procedure Calls.
+type HandlerFunc = Aeson.Value -> Stream -> IO (Either Text ())
+
+-- | notFoundHandlerFunc is tells the peer the Endpoint does not exist.
+notFoundHandlerFunc :: Endpoint -> HandlerFunc
+notFoundHandlerFunc endpoint _ _ = return
+  (Left $ "endpoint not found '" <> endpoint' <> "'")
+  where endpoint' = formatEndpoint endpoint
+
+-- | Handler can serve Remote Procedure Requests.
+class Handler h  where
+    endpoints
+        :: h
+        -> [Endpoint]
+
+    -- | Serve calls an incoming Remote Procedure Call.
+    serve
+        :: h
+        -> Endpoint
+        -> HandlerFunc
+
+    -- TODO: Maybe add 'conn' to notifyConnect
+
+    -- | notifyConnect tells the handler when a peer has connected.
+    notifyConnect
+        :: h
+        -> Ssb.PublicKey
+        -> IO (Either Text ())
+
+    -- | notifyDisconnect tells the handler when a peer has disconnected.
+    notifyDisconnect
+        :: h
+        -> Ssb.PublicKey
+        -> IO (Either Text ())
+
+-- | Router serves Remote Procedure Calls with a set of handlers.
+data Router = Router {
+      endpointHandlers :: Map Endpoint HandlerFunc
+    , connectCallbacks :: [Ssb.PublicKey -> IO (Either Text ())]
+    , disconnectCallbacks :: [Ssb.PublicKey -> IO (Either Text ())]
+    } deriving (Generic)
+
+instance Default Router
+
+instance Handler Router where
+  endpoints demuxer = Map.keys (endpointHandlers demuxer)
+
+  serve demuxer endpoint = case Map.lookup endpoint endpointHandlers' of
+    Nothing      -> (notFoundHandlerFunc endpoint)
+    Just handler -> handler
+    where endpointHandlers' = endpointHandlers demuxer
+
+  notifyConnect demuxer id = do
+    errs <- fmap lefts . sequence $ (\f -> f id) <$> connectCallbacks demuxer
+    if Protolude.null errs
+      then return . return $ ()
+      else return $ Left $ Text.intercalate ", " errs
+
+  notifyDisconnect demuxer id = do
+    errs <- fmap lefts . sequence $ (\f -> f id) <$> connectCallbacks demuxer
+    if Protolude.null errs
+      then return . return $ ()
+      else return $ Left $ Text.intercalate ", " errs
+
+
+-- | with adds a handler to the Router.
+with :: Handler h => Router -> h -> Router
+with demuxer handler = Router
+  { endpointHandlers    = Map.union (endpointHandlers demuxer) $ Map.fromList
+                            ((\e -> (e, serve handler e)) <$> endpoints handler)
+  , connectCallbacks    = connectCallbacks demuxer <> [notifyConnect handler]
+  , disconnectCallbacks = disconnectCallbacks demuxer
+                            <> [notifyDisconnect handler]
+  }
+
+-- | withM is a convenience function for using 'with' in monads.
+withM :: (MonadIO m, Handler h) => m Router -> m h -> m Router
+withM demuxer handler = with <$> demuxer <*> handler
+
+logMsg :: ConnState -> Text -> IO ()
+logMsg conn msg = do
+  _ <- atomically $ takeTMVar (lock conn)
+  print msg
+  atomically $ putTMVar (lock conn) True
+
+logDebug :: ConnState -> Text -> IO ()
+logDebug _ _ = return ()
+-- logDebug = logMsg
+
+-- | spawnConnection handles safely forking and closing RPC connections.
+spawnConnection
+  :: Stream
+  -> IO (Either Text ())
+  -> IO ()
+spawnConnection stream action = do
+  forkFinally
+    (do
+      res <- action
+      either (closeStreamWithError stream) (\_ -> closeStream stream) res
+    )
+    (\_ -> void (closeStream stream))
+  return ()
+
+-- | connect creates a Remote Procdure Call stream connection over the give Box
+-- Stream.
+connect
+  :: Handler h
+  => BoxStream.Conn
+  -> h
+  -> Ssb.PublicKey
+  -> (ConnState -> IO ())
+  -> IO (Either Text ())
+connect boxConn handler peer client = do
+  streamsIn          <- newTMVarIO Map.empty
+  streamsOut         <- newTMVarIO Map.empty
+  nextIncomingReqNum <- newTMVarIO 1
+  nextOutgoingReqNum <- newTMVarIO 1
+  lock               <- newTMVarIO True
+  let conn = ConnState { connPeer           = peer
+                       , boxConn            = boxConn
+                       , streamsIn          = streamsIn
+                       , streamsOut         = streamsOut
+                       , nextIncomingReqNum = nextIncomingReqNum
+                       , nextOutgoingReqNum = nextOutgoingReqNum
+                       , lock               = lock
+                       }
+
+  forkIO $
+    -- make RPC calls on the peer
+           client conn
+  print "entering service loop"
+  ret <- serviceLoop conn
+  print "out of service loop"
+  _   <- notifyDisconnect handler (connPeer conn)
+  print "disconnecting"
+  disconnect conn
+  print "disconnected"
+  return ret
+ where
+  serviceLoop conn = do
+    msg <- readMessage conn
+    case msg of
+      Left  err -> return $ error ("connection error: " <> err)
+      Right msg -> if header msg == goodByeHeader
+        then return . return $ ()
+        else do
+          ok <- handleMessage handler conn msg
+          if ok then serviceLoop conn else return . return $ ()
+
+-- TODO: Handle stream closing within demux loop
+--    The handleMessage loop is responsible for the lifetime of a stream
+--    connection.  It detects and creates requests, it should therefor handle
+--    close requests.
+--
+--    The stream type matters when closing a stream. Async do not require a
+--    specific close message.
+-- TODO: Handle closing of Async streams
+-- TODO: Properly handle Error message
+
+-- TODO: Refactor or reallocate helper functions
+--    These functions were added in a rush to get request handling working.  A
+--    better place could probably be found for them to improve code clarity.
+
+-- | isEndOfStream checks whether the message is the last of the stream.
+isEndOfStream = isEndOrError . flags . header
+
+-- | checks if the message is a response to an Async request.
+isAsyncResponse = not . isStream . flags . header
+
+-- | streamTable gets which lookup table to use for connections for the given
+-- stream.  There are seperate ones for incoming and outgoing RPC streams.
+streamTable stream = case direction stream of
+  Incoming -> streamsIn (conn stream)
+  Outgoing -> streamsOut (conn stream)
+
+streamStatus :: Stream -> STM StreamStatus
+streamStatus stream = do
+  table <- readTMVar $ streamTable stream
+  return
+    $ fromMaybe Closed (status . fst <$> Map.lookup (streamID stream) table)
+
+-- | manageStreamConn manages stream connection changes when writing messages
+-- to a stream.
+-- TODO: beware of entering a deadlock in manageStreamConn
+-- TODO: request stuff should be moved here
+manageStreamConn :: Stream -> Message -> IO (Either Text ())
+manageStreamConn stream msg = do
+  table <- atomically $ takeTMVar $ streamTable stream
+  res   <- writeMessage (conn stream) msg
+  let table' = if
+        | isLeft res -> closeStream table stream
+        | (not . isStream . flags . header $ msg) -> Map.adjust
+          (updateStreamStatus AwaitingResponse)
+          (streamID stream)
+          table
+        |
+-- TODO: confirm status of stream before closing it
+-- This happy path is not production ready
+          (isEndOrError . flags . header $ msg) -> Map.adjust
+          (updateStreamStatus AwaitingCloseRecv)
+          (streamID stream)
+          table
+        | otherwise -> table
+  when (isLeft res) $ do
+    let (_, c) = fromMaybe undefined $ Map.lookup (streamID stream) table
+    -- TODO: Find better way of closing channels on write error
+    atomically $ forM_ [1 .. 30] (\_ -> writeTChan c Nothing)
+  atomically $ putTMVar (streamTable stream) table'
+  return res
+ where
+  updateStreamStatus status (s, c) = (s { status = status }, c)
+  closeStream table tream = case direction stream of
+    Incoming -> Map.adjust (updateStreamStatus Closed) (streamID stream) table
+    Outgoing -> Map.delete (streamID stream) table
+
+
+-- TODO: Log important information for incoming messages
+handleMessage :: Handler h => h -> ConnState -> Message -> IO Bool
+handleMessage handler conn msg = do
+  nextReqNum <- atomically $ readTMVar (nextIncomingReqNum conn)
+  if
+    | (header msg) == goodByeHeader -> return False
+    | (isRequest nextReqNum msg) -> serveRequest handler conn msg
+    | (isEndOfStream msg) -> internalCloseStream conn msg
+    | (isAsyncResponse msg) -> do
+      demux conn msg
+      internalCloseStream conn msg
+    | otherwise -> demux conn msg
+ where
+  internalCloseStream conn msg = do
+    let flags          = def { isStream = True, isEndOrError = True }
+    let requestNumber' = (requestNumber . header $ msg)
+    let mVarTable =
+          if requestNumber' > 0 then streamsIn conn else streamsOut conn
+    table <- atomically $ takeTMVar mVarTable
+    let streamID'      = abs requestNumber'
+    let (stream, chan) = fromMaybe undefined $ Map.lookup streamID' table
+    newStatus <- case status stream of
+      Open -> do
+        let msg = fromRight undefined $ newCloseNotification streamID'
+        _ <- writeMessage conn msg
+        return AwaitingCloseRecv
+      AwaitingCloseRecv -> do
+        atomically $ writeTChan chan Nothing
+        return Closed
+      AwaitingResponse -> do
+        atomically $ writeTChan chan Nothing
+        return Closed
+      Closed -> do
+        logMsg conn
+          $  "received end of stream for already closed stream: "
+          <> (show streamID' :: Text)
+        return Closed
+    atomically $ putTMVar mVarTable $ case direction stream of
+      Incoming -> Map.adjust (updateStreamStatus newStatus) streamID' table
+      -- TODO: properly handle close of outgoing Duplex streams
+      Outgoing -> Map.delete streamID' table
+    return True
+    where updateStreamStatus status (s, c) = (s { status = status }, c)
+
+  serveRequest handler conn msg = do
+    let req = decodeJSONMessage msg
+    case req of
+      Left err -> do
+        logMsg conn $ "could not decode request: " <> err
+        return False
+      Right req -> do
+        let stream = Stream { streamID   = requestNumber . header $ msg
+                            , streamType = typ req
+                            , conn       = conn
+                            , direction  = Incoming
+                            , status     = Open
+                            , peer       = connPeer conn
+                            }
+        chan <- newTChanIO
+        err  <- atomically $ do
+          reqNum <- takeTMVar (nextIncomingReqNum conn)
+          table  <- takeTMVar (streamsIn conn)
+          if Map.size table == (maxBound :: Int)
+            then return $ Left errTooManyRequests
+            else do
+              let (reqNum', table') = if streamID stream == reqNum
+                    then (reqNum + 1, (Map.insert reqNum (stream, chan) table))
+                    else (reqNum, table)
+              putTMVar (nextIncomingReqNum conn) reqNum'
+              putTMVar (streamsIn conn)          table'
+              return . return $ ()
+        case err of
+          Left msg -> do
+            print msg
+            return False
+          Right _ -> do
+            -- Serving a request call, how do we close it nicely?
+            let endpoint = Endpoint (name req) (typ req)
+            spawnConnection stream $ (serve handler) endpoint (args req) stream
+            return True
+    where errTooManyRequests = "connection limit reached, dropping request"
+
+  demux conn msg = do
+    let reqNum    = requestNumber . header $ msg
+    let table = if reqNum > 0 then streamsIn conn else streamsOut conn
+    let streamID' = abs reqNum
+    table' <- atomically $ readTMVar table
+    case Map.lookup streamID' table' of
+      Nothing -> do
+        logDebug conn
+          $  "message dropped due to missing stream: "
+          <> (show msg :: Text)
+        return ()
+      Just (_, chan) -> atomically $ writeTChan chan (Just msg)
+    return True
+
+-- | disconnect is a hacked version, it rudely disconnects all connections in
+-- order to unlock streams stuck on their reading their channels.
+--
+-- Look into closing streams elegantly.
+-- TODO: Ensure disconnect prevents new incoming and outgoing connections
+disconnect :: ConnState -> IO ()
+disconnect conn = do
+  table <- atomically $ takeTMVar (streamsIn conn)
+  forM_ table
+    $ \(_, c) -> atomically $ forM_ [1 .. 30] (\_ -> writeTChan c Nothing)
+  let table' = Map.map (\(s, c) -> (s { status = Closed }, c)) table
+  atomically $ putTMVar (streamsIn conn) table'
+
+  table <- atomically $ takeTMVar (streamsOut conn)
+  forM_ table
+    $ \(_, c) -> atomically $ forM_ [1 .. 30] (\_ -> writeTChan c Nothing)
+  let table' = Map.map (\(s, c) -> (s { status = Closed }, c)) table
+  atomically $ putTMVar (streamsOut conn) table'
+
+  _ <- writeMessage conn $ Message goodByeHeader (BinaryPayload "")
+  return ()
+
+-- | readBoxStream reads the given number of bytes from the Box Stream.
+readBoxStream :: BoxStream.Conn -> Int -> IO (Either Text ByteString)
+readBoxStream conn bytes = do
+  mbuf <- BoxStream.readStream conn bytes
+  let buf = join $ (maybeToRight errUnexpectedClose) <$> mbuf
+  return buf
+  where errUnexpectedClose = "rpc.readConn: unexpected end of box stream"
+
+
+-- | readMessage reads a single message from the RPC connection.
+readMessage :: ConnState -> IO (Either Text Message)
+readMessage conn = do
+  headerBuf <- readBoxStream (boxConn conn) headerLength
+  header    <- (return $ headerBuf >>= decode)
+  case header of
+    Left  err    -> return $ Left $ "RPC.readMessage header: " <> err
+    Right header -> do
+      bodyBuf <- readBoxStream (boxConn conn) (fromIntegral $ bodyLength header)
+      let ret = liftA2 (<>) headerBuf bodyBuf >>= decode
+      logDebug conn
+        $  "readMessage ("
+        <> (Ssb.formatPublicKey $ connPeer conn)
+        <> ")"
+        <> (show ret)
+      return $ ret
+ where
+  decode :: Serialize a => ByteString -> Either Text a
+  decode = mapLeft toS . Serialize.decode
+
+writeMessage :: ConnState -> Message -> IO (Either Text ())
+writeMessage conn msg = do
+  logDebug conn
+    $  "writeMessage ("
+    <> (Ssb.formatPublicKey $ connPeer conn)
+    <> ")"
+    <> (show msg)
+  BoxStream.sendStream (boxConn conn) $ Serialize.encode msg
+
+-- | request makes a Remote Procedure Call on the peer.
+request
+  :: ToJSON a
+  => ConnState
+  -> Request a
+  -> (Stream -> IO (Either Text b))
+  -> IO (Either Text b)
+request conn req session = do
+  reqNum <- atomically $ takeTMVar (nextOutgoingReqNum conn)
+  let msg = newJSONMessage reqNum req
+  case msg of
+    Left err -> do
+      atomically $ putTMVar (nextOutgoingReqNum conn) reqNum
+      return $ Left err
+    Right msg -> do
+      let nextReqNum = reqNum + 1
+      atomically $ putTMVar (nextOutgoingReqNum conn) nextReqNum
+
+      -- TODO: Fix late night update
+      let flags' = flags . header $ msg
+      let
+        msg' = msg
+          { header = (header msg) { flags = flags'
+                                    { isStream = typ req /= Async
+                                    }
+                                  }
+          }
+
+      let stream = Stream { streamID   = reqNum
+                          , streamType = (typ req)
+                          , conn       = conn
+                          , direction  = Outgoing
+                          , status     = Open
+                          , peer       = connPeer conn
+                          }
+      atomically $ do
+        streams <- takeTMVar (streamsOut conn)
+        chan    <- newTChan
+        putTMVar (streamsOut conn) $ Map.insert reqNum (stream, chan) streams
+
+      _      <- writeMessage' stream msg'
+      result <- session stream
+      if streamType stream == Async
+        then return . return $ ()
+        else either (closeStreamWithError stream)
+                    (\x -> closeStream stream)
+                    result
+      return result
+  where writeMessage' = manageStreamConn
+
+-- | requestAsync is the Async version of request.
+requestAsync
+  :: (ToJSON a, FromJSON b) => ConnState -> Request a -> IO (Either Text b)
+requestAsync conn req = do
+  resp <- request conn req readStreamJSON
+  return $ resp >>= withErr "not response received"
+
+-- TODO: Use CloseNotification to signal close of stream.
+--   How is the JSON encoding done?
+data CloseNotification = CloseNotification ()
+
+newCloseNotification :: Int32 -> Either Text Message
+newCloseNotification streamID = do
+  let payload      = JSONPayload "true"
+  let requestNumer = -1 * streamID
+  header <- newHeader
+    (def { isEndOrError = True, isStream = True, bodyType = JSON })
+    requestNumer
+    "true"
+  return $ Message header payload
+
+-- | closeStream politely shuts down an RPC stream.
+-- TODO: closeStream needs to be more adaptive.  It only accounts for
+-- real streams.
+closeStream :: Stream -> IO (Either Text ())
+closeStream stream = case (status stream, streamType stream) of
+  (_   , Async) -> return . return $ ()
+  (Open, _    ) -> either (return . Left) (manageStreamConn stream)
+    $ newCloseNotification (streamID stream)
+  (Closed, _) -> return . return $ ()
+  (_     , _) -> return $ Left "could not close stream"
+
+-- ErrorNotification is the format used to communicate a stream error between peers.
+data ErrorNotification = ErrorNotification {
+      message :: Text
+    , stack   :: Maybe Text
+    }
+
+instance FromJSON ErrorNotification where
+  parseJSON = withObject "Error"
+    $ \v -> ErrorNotification <$> v .: "message" <*> v .: "stack"
+
+instance ToJSON ErrorNotification where
+  toJSON arg = object
+    [ "name" .= ("error" :: Text)
+    , "message" .= message arg
+    , "stack" .= stack arg
+    ]
+
+-- TODO: deduplicate closeStreamWithError and closeStream code
+-- TODO: Verify close operation on error
+--    Does the remote end send an 'true' message to confirm?  Or is the
+--    connection simply dropped.
+closeStreamWithError' :: Stream -> Text -> IO (Either Text ())
+closeStreamWithError' stream err = do
+  if streamType stream == Duplex || direction stream == Incoming
+    then do
+      let msg   = ErrorNotification err Nothing
+      let flags = def { isStream = True, isEndOrError = True }
+      err <- writeStream stream
+                         flags
+                         (JSONPayload $ BS.toStrict $ Aeson.encode msg)
+      either
+        (\err ->
+          logMsg (conn stream) $ "could not notify remote of error: " <> err
+        )
+        return
+        err
+    else logMsg (conn stream) $ "closing stream with error: " <> err
+  atomically $ do
+    let table = streamTable stream
+    value <- takeTMVar table
+    putTMVar table $ Map.delete (streamID stream) value
+  return . return $ ()
+ where
+  streamTable stream = case direction stream of
+    Incoming -> streamsIn $ conn stream
+    Outgoing -> streamsOut $ conn stream
+
+-- TODO: Merge newCloseErrorNotification with newCloseNotification It has
+-- proven difficult to use the ToJSON typeclass as an argument.
+newCloseErrorNotification :: Int32 -> Text -> Either Text Message
+newCloseErrorNotification streamID msg = do
+  let payload = JSONPayload $ BS.toStrict $ Aeson.encode msg
+  let requestNumer = -1 * streamID
+  header <- newHeader
+    (def { isEndOrError = True, isStream = True, bodyType = JSON })
+    requestNumer
+    "true"
+  return $ Message header payload
+
+closeStreamWithError :: Stream -> Text -> IO (Either Text ())
+closeStreamWithError stream err = case streamType stream of
+  Async -> return . return $ ()
+  otherwise ->
+    either (return . Left) (manageStreamConn stream)
+      $ newCloseErrorNotification (streamID stream) err
+
+-- TODO: Properly translate RPC errors
+-- TODO: Close stream within lock to avoid race conditions
+
+readStream :: Stream -> IO (Maybe MessagePayload)
+readStream stream = do
+  let table = streamTable stream
+  table' <- atomically $ readTMVar table
+  let result = Map.lookup (streamID stream) table'
+  msg <- case result of
+    Nothing               -> return Nothing
+    Just (stream', chan') -> if status stream' == Closed
+      then atomically $ join <$> tryReadTChan chan'
+      else atomically $ readTChan chan'
+  table'' <- atomically $ takeTMVar table
+  let table''' = case msg of
+        Nothing -> Map.delete (streamID stream) table'
+        Just _  -> table''
+  atomically $ putTMVar table table'''
+  return (body <$> msg)
+
+-- | readStreamJSON reads a single JSON message from the RPC stream.
+readStreamJSON :: FromJSON a => Stream -> IO (Either Text (Maybe a))
+readStreamJSON stream = do
+  resp <- readStream stream
+  case resp of
+    (Just (JSONPayload buf)) -> return (Just <$> decodeJSON buf)
+    (Just otherwise        ) -> return (errPayload resp)
+    Nothing                  -> return . return $ Nothing
+ where
+  errPayload payload =
+    error ("expected JSONPayload but got " <> (show payload :: Text))
+  errEOS = error "end of stream"
+
+writeStream :: Stream -> Flags -> MessagePayload -> IO (Either Text ())
+writeStream stream flags payload = do
+  status <- atomically $ streamStatus stream
+  if status /= Open
+    then return $ Left "stream closed"
+    else do
+      let msgID = case direction stream of
+            Incoming -> -1 * streamID stream
+            Outgoing -> streamID stream
+      let flags' = flags { isStream = not $ (streamType stream) == Async }
+      let header' = case payload of
+            (BinaryPayload p) ->
+              newHeader (flags' { bodyType = Binary }) msgID p
+            (TextPayload p) ->
+              newHeader (flags' { bodyType = UTF8String }) msgID p
+            (JSONPayload p) -> newHeader (flags' { bodyType = JSON }) msgID p
+      case header' of
+        Left  err     -> return $ Left err
+        Right header' -> do
+          let msg = Message header' payload
+          writeMessage' stream msg
+  where writeMessage' = manageStreamConn
+
+-- | writeStreamJSON writes a single JSON message to the RPC stream.
+writeStreamJSON :: ToJSON a => Stream -> a -> IO (Either Text ())
+writeStreamJSON stream msg = do
+  let buf = Aeson.encode msg
+  writeStream stream def (JSONPayload $ BS.toStrict buf)