@@ -25,7 +25,9 @@ import Streamly.Data.Fold as FL
2525import qualified Streamly.Internal.Data.Stream.IsStream as S
2626import qualified Data.Binary as B
2727import qualified Data.UUID as UUIDBase
28+import qualified Data.UUID.V4 as UUIDBase
2829import Control.Monad
30+import Control.Concurrent.STM
2931
3032import qualified Network.RPC.Curryer.StreamlyAdditions as SA
3133--import Control.Monad
@@ -88,41 +90,32 @@ deriving via WineryVariant Fingerprint instance Serialise Fingerprint
8890data MessageType = RequestMessage TimeoutMicroseconds
8991 | ResponseMessage
9092 | TimeoutResponseMessage
93+ | ExceptionResponseMessage
9194 deriving (Generic, Show)
9295 deriving Serialise via WineryVariant MessageType
9396
94−type MessageHandlers = [RequestHandler]
97+type MessageHandlers serverState = [RequestHandler serverState]
9598
96−--newtype RequestHandler = RequestHandler { handler :: forall a b. (Serialise a, Serialise b) => (a -> IO b) }
97−
98−data RequestHandler where
99+data RequestHandler serverState where
99100 -- | create a request handler with a response
100− RequestHandler :: forall a b. (Serialise a, Serialise b) => (a -> IO b) -> RequestHandler
101+ RequestHandler :: forall a b serverState. (Serialise a, Serialise b) => (ConnectionState serverState -> a -> IO b) -> RequestHandler serverState
101102 -- | create an asynchronous request handler where the client does not expect nor await a response
102− AsyncRequestHandler :: forall a. Serialise a => (a -> IO ()) -> RequestHandler
103+ AsyncRequestHandler :: forall a serverState. Serialise a => (ConnectionState serverState -> a -> IO ()) -> RequestHandler serverState
104+
105+-- | Passed to
106+data ConnectionState a = ConnectionState {
107+ connectionServerState :: STM a,
108+ connectionSocket :: Locking Socket
109+ }
110+
111+-- | Used by server-side request handlers to send additional messages to the client. This is useful for sending asynchronous responses to the client outside of the normal request-response flow.
112+sendMessage :: Serialise a => ConnectionState s -> a -> IO ()
113+sendMessage sState msg = do
114+ requestID <- UUID <$> UUIDBase.nextRandom
115+ let env =
116+ Envelope (fingerprint msg) (RequestMessage 0) requestID (serialise msg)
117+ sendEnvelope env (connectionSocket sState)
103118
104−{-
105−handleCall :: forall a b. (Serialise a, Serialise b)
106− => (a -> IO b)
107− -> RequestHandler
108−handleCall f = RequestHandler f
109−-}
110−{-data Message a = Response UUID a
111− | AsyncRequest UUID a
112− | ResponseExpectedRequest UUID (Maybe Timeout) a
113− | TimedOutResponse UUID
114− | ExceptionResponse UUID String
115− deriving (Generic, Show)
116− deriving Serialise via WineryVariant (Message a)
117−
118−-- | The response type
119−
120−data HandlerResponse a = NoResponse
121− | HandlerResponse a
122− | HandlerException String
123− deriving Generic
124− deriving Serialise via WineryVariant (HandlerResponse a)
125−-}
126119--avoid orphan instance
127120newtype UUID = UUID { _unUUID :: UUIDBase.UUID }
128121 deriving (Show, Eq, B.Binary, Hashable)
@@ -161,7 +154,8 @@ envelopeP = do
161154 msgTypeP = (P.satisfy (== 0) *>
162155 (RequestMessage . fromIntegral <$> word32P)) `P.alt`
163156 (P.satisfy (== 1) *> pure ResponseMessage) `P.alt`
164− (P.satisfy (== 2) *> pure TimeoutResponseMessage)
157+ (P.satisfy (== 2) *> pure TimeoutResponseMessage) `P.alt`
158+ (P.satisfy (== 3) *> pure ExceptionResponseMessage)
165159 lenPrefixedByteStringP = do
166160 c <- fromIntegral <$> word32P
167161 BS.pack <$> P.takeEQ c s
@@ -176,6 +170,7 @@ encodeEnvelope (Envelope (Fingerprint fp1 fp2) msgType msgId bs) =
176170 RequestMessage timeoutms -> BS.singleton 0 <> BO.bytestring32 (fromIntegral timeoutms)
177171 ResponseMessage -> BS.singleton 1
178172 TimeoutResponseMessage -> BS.singleton 2
173+ ExceptionResponseMessage -> BS.singleton 3
179174 msgIdBs =
180175 case UUIDBase.toWords (_unUUID msgId) of
181176 (u1, u2, u3, u4) -> foldr (<>) BS.empty (map BO.bytestring32 [u1, u2, u3, u4])
@@ -215,44 +210,34 @@ type NewConnectionHandler msg = IO (Maybe msg)
215210type NewMessageHandler req resp = req -> IO resp
216211
217212serve ::
218− MessageHandlers ->
213+ MessageHandlers s->
214+ STM s ->
219215 HostAddr ->
220216 PortNumber ->
221217 Maybe (MVar SockAddr) ->
222218 IO Bool
223−serve userMsgHandlers hostaddr port mSockLock = do
219+serve userMsgHandlers serverState hostaddr port mSockLock = do
224220 let
225221 handleSock sock = do
226222 lockingSocket <- newLock sock
227− drainSocketMessages sock (serverEnvelopeHandler lockingSocket userMsgHandlers)
223+ drainSocketMessages sock (serverEnvelopeHandler lockingSocket userMsgHandlers serverState)
228224
229225 S.serially (S.unfold (SA.acceptOnAddrWith [(ReuseAddr,1)] mSockLock) (hostaddr, port)) & S.parallely . S.mapM (handleWithM handleSock) & S.drain
230226 pure True
231227
232−{-serverEnvelopeHandler :: Locking Socket -> MessageHandlers s -> Envelope -> IO ()
233−serverEnvelopeHandler sockLock (Envelope fprint RequestMessage msgId messagebs) = do
234− error "spam"
235−serverEnvelopeHandler sockLock (Envelope fprint ResponseMessage msgId messagebs) = do
236− error "spam2"
237−-}
238−
239228openEnvelope :: forall s. (Serialise s, Typeable s) => Envelope -> Maybe s
240229openEnvelope (Envelope eprint _ _ bytes) = do
241− {-let decoder = case getDecoder (schema (Proxy :: Proxy s)) of
242− Left err -> error (show err)
243− Right dec -> dec-}
244230 if eprint == fingerprint (undefined :: s) then
245231 case deserialise bytes of
246232 Left err -> error (show err)
247233 Right v -> Just v
248− --Just (evalDecoder decoder bytes)
249234 else
250235 Nothing
251236
252−matchEnvelope :: forall a b. (Serialise a, Serialise b, Typeable b) =>
237+matchEnvelope :: forall a b s. (Serialise a, Serialise b, Typeable b) =>
253238 Envelope ->
254− (a -> IO b) ->
255− Maybe (a -> IO b, a)
239+ (ConnectionState s -> a -> IO b) ->
240+ Maybe (ConnectionState s -> a -> IO b, a)
256241matchEnvelope envelope dispatchf =
257242 case openEnvelope envelope :: Maybe a of
258243 Nothing -> Nothing
@@ -260,77 +245,54 @@ matchEnvelope envelope dispatchf =
260245
261246serverEnvelopeHandler ::
262247 Locking Socket
263− -> MessageHandlers
248+ -> MessageHandlers s
249+ -> STM s
264250 -> Envelope
265251 -> IO ()
266−serverEnvelopeHandler _ _ (Envelope _ TimeoutResponseMessage _ _) = pure ()
267−serverEnvelopeHandler sockLock msgHandlers envelope@(Envelope _ (RequestMessage timeoutms) msgId _) = do
252+serverEnvelopeHandler _ _ _ (Envelope _ TimeoutResponseMessage _ _) = pure ()
253+serverEnvelopeHandler _ _ _ (Envelope _ ExceptionResponseMessage _ _) = pure ()
254+serverEnvelopeHandler sockLock msgHandlers serverState envelope@(Envelope _ (RequestMessage timeoutms) msgId _) = do
268255 --find first matching handler
269256 let runTimeout :: IO b -> IO (Maybe b)
270− runTimeout m =
257+ runTimeout m =
271258 if timeoutms == 0 then
272259 Just <$> m
273− else
274− timeout (fromIntegral timeoutms) m
275− firstMatcher (RequestHandler (msghandler :: a -> IO b)) Nothing =
260+ else
261+ timeout (fromIntegral timeoutms) m
262+
263+ sState = ConnectionState {
264+ connectionServerState = serverState,
265+ connectionSocket = sockLock
266+ }
267+
268+ firstMatcher (RequestHandler msghandler) Nothing =
276269 case matchEnvelope envelope msghandler of
277270 Nothing -> pure Nothing
278− Just (dispatchf, decoded) ->
271+ Just (dispatchf, decoded) -> do
279272 --TODO add exception handling
280− do
281− mResponse <- runTimeout (dispatchf decoded)
282− let envelopeResponse =
283− case mResponse of
284− Just response ->
285− Envelope (fingerprint response) ResponseMessage msgId (serialise response)
286− Nothing -> Envelope (fingerprint TimeoutError) (TimeoutResponseMessage) msgId (BS.empty)
287− sendEnvelope envelopeResponse sockLock
288− pure (Just ())
289− firstMatcher (AsyncRequestHandler (msghandler :: a -> IO ())) Nothing =
273+ mResponse <- runTimeout (dispatchf sState decoded)
274+ let envelopeResponse =
275+ case mResponse of
276+ Just response ->
277+ Envelope (fingerprint response) ResponseMessage msgId (serialise response)
278+ Nothing ->
279+ Envelope (fingerprint TimeoutError) (TimeoutResponseMessage) msgId (BS.empty)
280+ sendEnvelope envelopeResponse sockLock
281+ pure (Just ())
282+ firstMatcher (AsyncRequestHandler msghandler) Nothing =
290283 case matchEnvelope envelope msghandler of
291284 Nothing -> pure Nothing
292285 Just (dispatchf, decoded) -> do
293− _ <- dispatchf decoded
286+ _ <- dispatchf sState decoded
294287 pure (Just ())
295288 firstMatcher _ acc = pure acc
296− foldM_ (flip firstMatcher) Nothing msgHandlers
297−serverEnvelopeHandler _ _ (Envelope _ ResponseMessage _ _) = error "server received response message"
298−
299−
300−{-
301−
302−
303− --putStrLn $ "GOT MSG " ++ show msg
304− let --runTimeout :: IO (HandlerResponse resp) -> IO (Maybe (HandlerResponse resp))
305− runTimeout mTimeout m = case mTimeout of
306− Nothing -> Just <$> m
307− Just timeoutMicroseconds -> timeout timeoutMicroseconds m
308− case msg of
309− Response{} -> putStrLn "client sent response"
310− ResponseExpectedRequest requestID mTimeout val -> do
311− let normalResponder = do
312− resp <- runTimeout mTimeout (requestMessageHandler val)
313− case resp of
314− Just (HandlerResponse responseVal) -> sendMessage (Response requestID responseVal) sockLock
315− Just NoResponse -> error "attempt to return non-response to expected response message"
316− Just (HandlerException exc) -> sendMessage (ExceptionResponse @(Message resp) requestID exc) sockLock
317− Nothing ->
318− sendMessage (TimedOutResponse @(Message resp) requestID) sockLock
319− excHandler :: SomeException -> IO ()
320− excHandler e = do
321− --send exception to client
322− sendMessage (ExceptionResponse @(Message resp) requestID (show e)) sockLock
323− throwIO e
324− catch normalResponder excHandler
325− AsyncRequest _ val ->
326− void $ requestMessageHandler val
327− --no response necessary
328− ExceptionResponse{} -> putStrLn "client sent exception response"
329− TimedOutResponse{} -> putStrLn "client sent timed out response"
330−-}
331−
332−--add callback to allow for responses via socket
333−--type MessageHandler a = Message a -> IO ()
289+ eExc <- try $ foldM_ (flip firstMatcher) Nothing msgHandlers :: IO (Either SomeException ())
290+ case eExc of
291+ Left exc ->
292+ let env = Envelope (fingerprint (show exc)) ExceptionResponseMessage msgId (serialise (show exc)) in
293+ sendEnvelope env sockLock
294+ Right () -> pure ()
295+serverEnvelopeHandler _ _ _ (Envelope _ ResponseMessage _ _) = error "server received response message"
334296
335297type EnvelopeHandler = Envelope -> IO ()
336298