@@ -1,7 +1,8 @@
1−{-# LANGUAGE DerivingVia, DeriveGeneric, RankNTypes, ScopedTypeVariables, MultiParamTypeClasses, OverloadedStrings, GeneralizedNewtypeDeriving, CPP, ExistentialQuantification, StandaloneDeriving, GADTs, RecordWildCards #-}
1+{-# LANGUAGE DerivingVia, DeriveGeneric, RankNTypes, ScopedTypeVariables, MultiParamTypeClasses, OverloadedStrings, GeneralizedNewtypeDeriving, TypeApplications, CPP, ExistentialQuantification, StandaloneDeriving #-}
22{-# OPTIONS_GHC -fno-warn-orphans #-}
33{- HLINT ignore "Use lambda-case" -}
44module Network.RPC.Curryer.Server where
5+import Streamly
56import qualified Streamly.Prelude as S
67import Streamly.Network.Socket
78import Streamly.Internal.Network.Socket (handleWithM)
@@ -25,24 +26,21 @@ import Streamly.Data.Fold as FL
2526import qualified Streamly.Internal.Data.Stream.IsStream as S
2627import qualified Data.Binary as B
2728import qualified Data.UUID as UUIDBase
28−import qualified Data.UUID.V4 as UUIDBase
29−import Control.Monad
30−import Data.Functor
3129
3230import qualified Network.RPC.Curryer.StreamlyAdditions as SA
3331--import Control.Monad
3432import Data.Hashable
35−import System.Timeout
33+--import System.Timeout
3634import qualified Network.ByteOrder as BO
3735
38−
3936-- for toArrayS conversion
40−import qualified Data.ByteString.Internal as BSI
41−import qualified Streamly.Internal.Data.Array.Storable.Foreign.Types as Arr
37+{-import qualified Data.ByteString.Internal as BSI
38+import qualified Streamly.Internal.Data.Array.Storable.Foreign.Types as SA
39+import Foreign.ForeignPtr (plusForeignPtr)
4240import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
43−import GHC.Ptr (minusPtr)
44−
45−-- define CURRYER_SHOW_BYTES 1
41+import GHC.Ptr (plusPtr)
42+-}
43+-- #define CURRYER_SHOW_BYTES 1
4644#if CURRYER_SHOW_BYTES == 1
4745import Debug.Trace
4846#endif
@@ -68,58 +66,45 @@ withLock (Locking mvar v) m =
6866lockless :: Locking a -> a
6967lockless (Locking _ a) = a
7068
71−type Timeout = Word32
69+type Timeout = Int
7270
7371type BinaryMessage = BS.ByteString
7472
7573--includes the fingerprint of the incoming data type (held in the BinaryMessage) to determine how to dispatch the message.
7674--add another envelope type for unencoded binary messages for any easy optimization for in-process communication
77−data Envelope = Envelope {
78− envFingerprint :: !Fingerprint,
79− envMessageType :: !MessageType,
80− envMsgId :: !UUID,
81− envPayload :: !BinaryMessage
82− }
75+data Envelope = Envelope !Fingerprint !MessageType !UUID !BinaryMessage
8376 deriving (Generic, Show)
84−
85−type TimeoutMicroseconds = Int
77+ deriving Serialise via WineryVariant Envelope
8678
8779deriving instance Generic Fingerprint
8880deriving via WineryVariant Fingerprint instance Serialise Fingerprint
8981
90−-- | Internal type used to mark envelope types.
91−data MessageType = RequestMessage TimeoutMicroseconds
82+data MessageType = RequestMessage
9283 | ResponseMessage
93− | TimeoutResponseMessage
94− | ExceptionResponseMessage
9584 deriving (Generic, Show)
9685 deriving Serialise via WineryVariant MessageType
9786
98−-- | A list of `RequestHandler`s.
99−type RequestHandlers serverState = [RequestHandler serverState]
100−
101−-- | Data types for server-side request handlers, in synchronous (client waits for return value) and asynchronous (client does not wait for return value) forms.
102−data RequestHandler serverState where
103− -- | create a request handler with a response
104− RequestHandler :: forall a b serverState. (Serialise a, Serialise b) => (ConnectionState serverState -> a -> IO b) -> RequestHandler serverState
105− -- | create an asynchronous request handler where the client does not expect nor await a response
106− AsyncRequestHandler :: forall a serverState. Serialise a => (ConnectionState serverState -> a -> IO ()) -> RequestHandler serverState
107−
108−-- | Server state sent in via `serve` and passed to `RequestHandler`s.
109−data ConnectionState a = ConnectionState {
110− connectionServerState :: a,
111− connectionSocket :: Locking Socket
112− }
113−
114−-- | 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. The locking socket can be found in the ConnectionState when a request handler is called.
115−sendMessage :: Serialise a => Locking Socket -> a -> IO ()
116−sendMessage lockSock msg = do
117− requestID <- UUID <$> UUIDBase.nextRandom
118− let env =
119− Envelope (fingerprint msg) (RequestMessage timeout') requestID (serialise msg)
120− timeout' = 0
121− sendEnvelope env lockSock
87+data MessageHandlers =
88+ forall a b. (Show a, Serialise a, Serialise b) => MessageHandlers [RequestHandler a b]
12289
90+type RequestHandler a b = a -> IO b
91+
92+{-data Message a = Response UUID a
93+ | AsyncRequest UUID a
94+ | ResponseExpectedRequest UUID (Maybe Timeout) a
95+ | TimedOutResponse UUID
96+ | ExceptionResponse UUID String
97+ deriving (Generic, Show)
98+ deriving Serialise via WineryVariant (Message a)
99+
100+-- | The response type
101+
102+data HandlerResponse a = NoResponse
103+ | HandlerResponse a
104+ | HandlerException String
105+ deriving Generic
106+ deriving Serialise via WineryVariant (HandlerResponse a)
107+-}
123108--avoid orphan instance
124109newtype UUID = UUID { _unUUID :: UUIDBase.UUID }
125110 deriving (Show, Eq, B.Binary, Hashable)
@@ -138,18 +123,13 @@ instance Serialise UUID where
138123 x -> error $ "invalid schema element " <> show x
139124 decodeCurrent = B.decode . BSL.fromStrict <$> (decodeVarInt >>= getBytes)
140125
141−-- | Errors from remote calls.
126+
142127data ConnectionError = CodecError String -- show of WineryException from exception initiator which cannot otherwise be transmitted over a line due to dependencies on TypeReps
143128 | TimeoutError
144129 | ExceptionError String
145130 deriving (Generic, Show, Eq)
146131 deriving Serialise via WineryVariant ConnectionError
147132
148−data TimeoutException = TimeoutException
149− deriving Show
150−
151−instance Exception TimeoutException
152−
153133type HostAddr = (Word8, Word8, Word8, Word8)
154134
155135allHostAddrs,localHostAddr :: HostAddr
@@ -159,18 +139,12 @@ localHostAddr = (127,0,0,1)
159139-- Each message is length-prefixed by a 32-bit unsigned length.
160140envelopeP :: Parser IO Word8 Envelope
161141envelopeP = do
162− let msgTypeP = (P.satisfy (== 0) *>
163− (RequestMessage . fromIntegral <$> word32P)) `P.alt`
164− (P.satisfy (== 1) $> ResponseMessage) `P.alt`
165− (P.satisfy (== 2) $> TimeoutResponseMessage) `P.alt`
166− (P.satisfy (== 3) $> ExceptionResponseMessage)
142+ let s = FL.toList
143+ msgTypeP = (P.satisfy (== 0) *> pure RequestMessage) `P.alt`
144+ (P.satisfy (== 1) *> pure ResponseMessage)
167145 lenPrefixedByteStringP = do
168146 c <- fromIntegral <$> word32P
169− --streamly can't handle takeEQ 0, so add special handling
170− if c == 0 then
171− pure BS.empty
172− else
173− fromArray <$> P.takeEQ c (Arr.writeN c)
147+ BS.pack <$> P.takeEQ c s
174148 Envelope <$> fingerprintP <*> msgTypeP <*> uuidP <*> lenPrefixedByteStringP
175149
176150encodeEnvelope :: Envelope -> BS.ByteString
@@ -179,19 +153,19 @@ encodeEnvelope (Envelope (Fingerprint fp1 fp2) msgType msgId bs) =
179153 where
180154 fingerprintBs = BO.bytestring64 fp1 <> BO.bytestring64 fp2
181155 msgTypeBs = case msgType of
182− RequestMessage timeoutms -> BS.singleton 0 <> BO.bytestring32 (fromIntegral timeoutms)
156+ RequestMessage -> BS.singleton 0
183157 ResponseMessage -> BS.singleton 1
184− TimeoutResponseMessage -> BS.singleton 2
185− ExceptionResponseMessage -> BS.singleton 3
186158 msgIdBs =
187159 case UUIDBase.toWords (_unUUID msgId) of
188− (u1, u2, u3, u4) -> foldr ((<>) . BO.bytestring32) BS.empty [u1, u2, u3, u4]
160+ (u1, u2, u3, u4) -> foldr (<>) BS.empty (map BO.bytestring32 [u1, u2, u3, u4])
189161 msgLen = fromIntegral (BS.length bs)
190162 lenPrefixedBs = BO.bytestring32 msgLen <> bs
191163
192164fingerprintP :: Parser IO Word8 Fingerprint
193−fingerprintP =
194− Fingerprint <$> word64P <*> word64P
165+fingerprintP = do
166+ f1 <- word64P
167+ f2 <- word64P
168+ pure (Fingerprint f1 f2)
195169
196170word64P :: Parser IO Word8 Word64
197171word64P = do
@@ -212,112 +186,112 @@ uuidP = do
212186 u1 <- word32P
213187 u2 <- word32P
214188 u3 <- word32P
215− --u4 <- word32P
216− --pure (UUID (UUIDBase.fromWords u1 u2 u3 u4))-}
217− --(UUID . UUIDBase.fromWords) <$> word32P <*> word32P <*> word32P <*> word32P
218− UUID . UUIDBase.fromWords u1 u2 u3 <$> word32P
189+ u4 <- word32P
190+ pure (UUID (UUIDBase.fromWords u1 u2 u3 u4))
219191
220192type NewConnectionHandler msg = IO (Maybe msg)
221193
222194type NewMessageHandler req resp = req -> IO resp
223−
224−-- | Listen for new connections and handle requests which are passed the server state 's'. The MVar SockAddr can be be optionally used to know when the server is ready for processing requests.
195+
225196serve ::
226− RequestHandlers s->
227− s ->
197+ MessageHandlers ->
228198 HostAddr ->
229199 PortNumber ->
230200 Maybe (MVar SockAddr) ->
231201 IO Bool
232−serve userMsgHandlers serverState hostaddr port mSockLock = do
202+serve userMsgHandler hostaddr port mSockLock = do
233203 let
234204 handleSock sock = do
235205 lockingSocket <- newLock sock
236− drainSocketMessages sock (serverEnvelopeHandler lockingSocket userMsgHandlers serverState)
206+ drainSocketMessages sock (serverEnvelopeHandler lockingSocket userMsgHandler)
237207
238− S.serially (S.unfold (SA.acceptOnAddrWith [(ReuseAddr,1)] mSockLock) (hostaddr, port)) & S.parallely . S.mapM (handleWithM handleSock) & S.drain
208+ serially (S.unfold (SA.acceptOnAddrWith [(ReuseAddr,1)] mSockLock) (hostaddr, port)) & parallely . S.mapM (handleWithM handleSock) & S.drain
239209 pure True
240210
211+{-serverEnvelopeHandler :: Locking Socket -> MessageHandlers s -> Envelope -> IO ()
212+serverEnvelopeHandler sockLock (Envelope fprint RequestMessage msgId messagebs) = do
213+ error "spam"
214+serverEnvelopeHandler sockLock (Envelope fprint ResponseMessage msgId messagebs) = do
215+ error "spam2"
216+-}
217+
241218openEnvelope :: forall s. (Serialise s, Typeable s) => Envelope -> Maybe s
242−openEnvelope (Envelope eprint _ _ bytes) =
219+openEnvelope (Envelope eprint _ _ bytes) = do
220+ {-let decoder = case getDecoder (schema (Proxy :: Proxy s)) of
221+ Left err -> error (show err)
222+ Right dec -> dec-}
243223 if eprint == fingerprint (undefined :: s) then
244224 case deserialise bytes of
245225 Left err -> error (show err)
246226 Right v -> Just v
227+ --Just (evalDecoder decoder bytes)
247228 else
248229 Nothing
249230
250−matchEnvelope :: forall a b s. (Serialise a, Serialise b, Typeable b) =>
251− Envelope ->
252− (ConnectionState s -> a -> IO b) ->
253− Maybe (ConnectionState s -> a -> IO b, a)
254−matchEnvelope envelope dispatchf =
255− case openEnvelope envelope :: Maybe a of
256− Nothing -> Nothing
257− Just decoded -> Just (dispatchf, decoded)
258−
259−-- | Called by `serve` to process incoming envelope requests. Never returns, so use `async` to spin it off on another thread.
260231serverEnvelopeHandler ::
261232 Locking Socket
262− -> RequestHandlers s
263− -> s
233+ -> MessageHandlers
264234 -> Envelope
265235 -> IO ()
266−serverEnvelopeHandler _ _ _ (Envelope _ TimeoutResponseMessage _ _) = pure ()
267−serverEnvelopeHandler _ _ _ (Envelope _ ExceptionResponseMessage _ _) = pure ()
268−serverEnvelopeHandler _ _ _ (Envelope _ ResponseMessage _ _) = pure ()
269−serverEnvelopeHandler sockLock msgHandlers serverState envelope@(Envelope _ (RequestMessage timeoutms) msgId _) = do
236+serverEnvelopeHandler sockLock (MessageHandlers msgHandlers) envelope@(Envelope _ RequestMessage msgId _) = do
270237 --find first matching handler
271− let runTimeout :: IO b -> IO (Maybe b)
272− runTimeout m =
273− if timeoutms == 0 then
274− (Just <$> m) `catch` timeoutExcHandler
275− else
276− (timeout (fromIntegral timeoutms) m) `catch` timeoutExcHandler
277− --allow server-side function to throw TimeoutError which is caught here and becomes TimeoutError value
278− timeoutExcHandler :: TimeoutException -> IO (Maybe b)
279− timeoutExcHandler _ = pure Nothing
238+ let firstMatch = foldr firstMatcher Nothing msgHandlers
239+ firstMatcher :: forall a b. (Serialise a, Serialise b) => RequestHandler a b -> Maybe (RequestHandler a b, a) -> Maybe (RequestHandler a b, a)
240+ firstMatcher msghandler Nothing =
241+ case openEnvelope envelope :: Maybe a of
242+ Just decoded -> Just (msghandler, decoded)
243+ Nothing -> Nothing
244+ firstMatcher _ acc = acc
245+ case firstMatch of
246+ Nothing -> error "failed to handle msg"
247+ Just (msghandler, decoded) -> do
248+ --TODO add exception handling
249+ response <- msghandler decoded
250+ sendEnvelope (Envelope (fingerprint response) ResponseMessage msgId (serialise response)) sockLock
251+serverEnvelopeHandler _ _ (Envelope _ ResponseMessage _ _) = error "server received response message"
252+{-
280253
281− sState = ConnectionState {
282− connectionServerState = serverState,
283− connectionSocket = sockLock
284− }
285−
286− firstMatcher (RequestHandler msghandler) Nothing =
287− case matchEnvelope envelope msghandler of
288− Nothing -> pure Nothing
289− Just (dispatchf, decoded) -> do
290− --TODO add exception handling
291− mResponse <- runTimeout (dispatchf sState decoded)
292− let envelopeResponse =
293− case mResponse of
294− Just response ->
295− Envelope (fingerprint response) ResponseMessage msgId (serialise response)
296− Nothing ->
297− Envelope (fingerprint TimeoutError) TimeoutResponseMessage msgId BS.empty
298− sendEnvelope envelopeResponse sockLock
299− pure (Just ())
300− firstMatcher (AsyncRequestHandler msghandler) Nothing =
301− case matchEnvelope envelope msghandler of
302− Nothing -> pure Nothing
303− Just (dispatchf, decoded) -> do
304− _ <- dispatchf sState decoded
305− pure (Just ())
306− firstMatcher _ acc = pure acc
307− eExc <- try $ foldM_ (flip firstMatcher) Nothing msgHandlers :: IO (Either SomeException ())
308− case eExc of
309− Left exc ->
310− let env = Envelope (fingerprint (show exc)) ExceptionResponseMessage msgId (serialise (show exc)) in
311− sendEnvelope env sockLock
312− Right () -> pure ()
313−
254+
255+ --putStrLn $ "GOT MSG " ++ show msg
256+ let --runTimeout :: IO (HandlerResponse resp) -> IO (Maybe (HandlerResponse resp))
257+ runTimeout mTimeout m = case mTimeout of
258+ Nothing -> Just <$> m
259+ Just timeoutMicroseconds -> timeout timeoutMicroseconds m
260+ case msg of
261+ Response{} -> putStrLn "client sent response"
262+ ResponseExpectedRequest requestID mTimeout val -> do
263+ let normalResponder = do
264+ resp <- runTimeout mTimeout (requestMessageHandler val)
265+ case resp of
266+ Just (HandlerResponse responseVal) -> sendMessage (Response requestID responseVal) sockLock
267+ Just NoResponse -> error "attempt to return non-response to expected response message"
268+ Just (HandlerException exc) -> sendMessage (ExceptionResponse @(Message resp) requestID exc) sockLock
269+ Nothing ->
270+ sendMessage (TimedOutResponse @(Message resp) requestID) sockLock
271+ excHandler :: SomeException -> IO ()
272+ excHandler e = do
273+ --send exception to client
274+ sendMessage (ExceptionResponse @(Message resp) requestID (show e)) sockLock
275+ throwIO e
276+ catch normalResponder excHandler
277+ AsyncRequest _ val ->
278+ void $ requestMessageHandler val
279+ --no response necessary
280+ ExceptionResponse{} -> putStrLn "client sent exception response"
281+ TimedOutResponse{} -> putStrLn "client sent timed out response"
282+-}
283+
284+--add callback to allow for responses via socket
285+--type MessageHandler a = Message a -> IO ()
314286
315287type EnvelopeHandler = Envelope -> IO ()
316288
289+type AsyncMessageHandler a = a -> IO ()
290+
317291drainSocketMessages :: Socket -> EnvelopeHandler -> IO ()
318292drainSocketMessages sock envelopeHandler = do
319293 let sockStream = S.unfold readWithBufferOf (1024 * 4, sock)
320− S.drain $ S.serially $ S.parseMany envelopeP sockStream & S.mapM envelopeHandler
294+ S.drain $ serially $ S.parseMany envelopeP sockStream & S.mapM envelopeHandler
321295
322296--send length-tagged bytestring, perhaps should be in network byte order?
323297sendEnvelope :: Envelope -> Locking Socket -> IO ()
@@ -327,16 +301,7 @@ sendEnvelope envelope sockLock = do
327301 --Socket.sendAll syscalls send() on a loop until all the bytes are sent, so we need socket locking here to account for serialized messages of size > PIPE_BUF
328302 withLock sockLock $ \socket' ->
329303 Socket.sendAll socket' fullbytes
330− --traceShowM ("sendEnvelope", envelope)
331304 traceBytes "sendEnvelope" fullbytes
332305
333306fingerprint :: Typeable a => a -> Fingerprint
334307fingerprint = typeRepFingerprint . typeOf
335−
336−fromArray :: Arr.Array Word8 -> BSI.ByteString
337−fromArray Arr.Array {..}
338− | aLen == 0 = mempty
339− | otherwise = BSI.PS aStart 0 aLen
340− where
341− aStartPtr = unsafeForeignPtrToPtr aStart
342− aLen = aEnd `minusPtr` aStartPtr