@@ -1,4 +1,5 @@
1−{-# LANGUAGE DerivingVia, DeriveGeneric, RankNTypes, ScopedTypeVariables, MultiParamTypeClasses, OverloadedStrings, GeneralizedNewtypeDeriving, TypeApplications, CPP #-}
1+{-# LANGUAGE DerivingVia, DeriveGeneric, RankNTypes, ScopedTypeVariables, MultiParamTypeClasses, OverloadedStrings, GeneralizedNewtypeDeriving, TypeApplications, CPP, ExistentialQuantification, StandaloneDeriving #-}
2+{-# OPTIONS_GHC -fno-warn-orphans #-}
23{- HLINT ignore "Use lambda-case" -}
34module Network.RPC.Curryer.Server where
45import Streamly
@@ -7,11 +8,13 @@ import Streamly.Network.Socket
78import Streamly.Internal.Network.Socket (handleWithM)
89import Network.Socket as Socket
910import Network.Socket.ByteString as Socket
10−import Streamly.Internal.Data.Parser as P
11+import Streamly.Internal.Data.Parser as P hiding (concatMap)
1112import Codec.Winery
1213import Codec.Winery.Internal (varInt, decodeVarInt, getBytes)
1314import Codec.Winery.Class (mkExtractor)
1415import GHC.Generics
16+import GHC.Fingerprint
17+import Data.Typeable
1518import Control.Concurrent.MVar (MVar, newMVar, withMVar)
1619import Control.Exception
1720import Data.Function ((&))
@@ -23,14 +26,12 @@ import Streamly.Data.Fold as FL
2326import qualified Streamly.Internal.Data.Stream.IsStream as S
2427import qualified Data.Binary as B
2528import qualified Data.UUID as UUIDBase
26−import qualified Data.UUID.V4 as UUIDBase
2729
2830import qualified Network.RPC.Curryer.StreamlyAdditions as SA
29−import Control.Monad
31+--import Control.Monad
3032import Data.Hashable
31−import System.Timeout
33+--import System.Timeout
3234import qualified Network.ByteOrder as BO
33−import Data.Proxy
3435
3536-- for toArrayS conversion
3637{-import qualified Data.ByteString.Internal as BSI
@@ -39,6 +40,7 @@ import Foreign.ForeignPtr (plusForeignPtr)
3940import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
4041import GHC.Ptr (plusPtr)
4142-}
43+-- #define CURRYER_SHOW_BYTES 1
4244#if CURRYER_SHOW_BYTES == 1
4345import Debug.Trace
4446#endif
@@ -66,7 +68,28 @@ lockless (Locking _ a) = a
6668
6769type Timeout = Int
6870
69−data Message a = Response UUID a
71+type BinaryMessage = BS.ByteString
72+
73+--includes the fingerprint of the incoming data type (held in the BinaryMessage) to determine how to dispatch the message.
74+--add another envelope type for unencoded binary messages for any easy optimization for in-process communication
75+data Envelope = Envelope !Fingerprint !MessageType !UUID !BinaryMessage
76+ deriving (Generic, Show)
77+ deriving Serialise via WineryVariant Envelope
78+
79+deriving instance Generic Fingerprint
80+deriving via WineryVariant Fingerprint instance Serialise Fingerprint
81+
82+data MessageType = RequestMessage
83+ | ResponseMessage
84+ deriving (Generic, Show)
85+ deriving Serialise via WineryVariant MessageType
86+
87+data MessageHandlers =
88+ forall a b. (Show a, Serialise a, Serialise b) => MessageHandlers [RequestHandler a b]
89+
90+type RequestHandler a b = a -> IO b
91+
92+{-data Message a = Response UUID a
7093 | AsyncRequest UUID a
7194 | ResponseExpectedRequest UUID (Maybe Timeout) a
7295 | TimedOutResponse UUID
@@ -75,12 +98,13 @@ data Message a = Response UUID a
7598 deriving Serialise via WineryVariant (Message a)
7699
77100-- | The response type
101+
78102data HandlerResponse a = NoResponse
79103 | HandlerResponse a
80104 | HandlerException String
81105 deriving Generic
82106 deriving Serialise via WineryVariant (HandlerResponse a)
83−
107+-}
84108--avoid orphan instance
85109newtype UUID = UUID { _unUUID :: UUIDBase.UUID }
86110 deriving (Show, Eq, B.Binary, Hashable)
@@ -113,56 +137,121 @@ allHostAddrs = (0,0,0,0)
113137localHostAddr = (127,0,0,1)
114138
115139-- Each message is length-prefixed by a 32-bit unsigned length.
116−messageBoundaryP :: Parser IO Word8 BS.ByteString
117−messageBoundaryP = do
140+envelopeP :: Parser IO Word8 Envelope
141+envelopeP = do
142+ let s = FL.toList
143+ msgTypeP = (P.satisfy (== 0) *> pure RequestMessage) `P.alt`
144+ (P.satisfy (== 1) *> pure ResponseMessage)
145+ lenPrefixedByteStringP = do
146+ c <- fromIntegral <$> word32P
147+ BS.pack <$> P.takeEQ c s
148+ Envelope <$> fingerprintP <*> msgTypeP <*> uuidP <*> lenPrefixedByteStringP
149+
150+encodeEnvelope :: Envelope -> BS.ByteString
151+encodeEnvelope (Envelope (Fingerprint fp1 fp2) msgType msgId bs) =
152+ fingerprintBs <> msgTypeBs <> msgIdBs <> lenPrefixedBs
153+ where
154+ fingerprintBs = BO.bytestring64 fp1 <> BO.bytestring64 fp2
155+ msgTypeBs = case msgType of
156+ RequestMessage -> BS.singleton 0
157+ ResponseMessage -> BS.singleton 1
158+ msgIdBs =
159+ case UUIDBase.toWords (_unUUID msgId) of
160+ (u1, u2, u3, u4) -> foldr (<>) BS.empty (map BO.bytestring32 [u1, u2, u3, u4])
161+ msgLen = fromIntegral (BS.length bs)
162+ lenPrefixedBs = BO.bytestring32 msgLen <> bs
163+
164+fingerprintP :: Parser IO Word8 Fingerprint
165+fingerprintP = do
166+ f1 <- word64P
167+ f2 <- word64P
168+ pure (Fingerprint f1 f2)
169+
170+word64P :: Parser IO Word8 Word64
171+word64P = do
118172 let s = FL.toList
119− w4x8 <- P.take 4 s
120− --traceShowM ("w4x8"::String, w4x8)
121− let c = fromIntegral (BO.word32 (BS.pack w4x8))
122− --traceShowM ("c"::String, c)
123− vals <- P.take c s
124− let bytes = BS.pack vals
125− --traceShowM ("parsedBytes"::String, c, BS.length bytes, bytes)
126− pure bytes
173+ b <- P.takeEQ 8 s
174+ pure (BO.word64 (BS.pack b))
175+
176+--parse a 32-bit integer from network byte order
177+word32P :: Parser IO Word8 Word32
178+word32P = do
179+ let s = FL.toList
180+ w4x8 <- P.takeEQ 4 s
181+ pure (BO.word32 (BS.pack w4x8))
182+
183+-- uuid is encode as 4 32-bit words because of its convenient 32-bit tuple encoding
184+uuidP :: Parser IO Word8 UUID
185+uuidP = do
186+ u1 <- word32P
187+ u2 <- word32P
188+ u3 <- word32P
189+ u4 <- word32P
190+ pure (UUID (UUIDBase.fromWords u1 u2 u3 u4))
127191
128192type NewConnectionHandler msg = IO (Maybe msg)
129193
130−type NewMessageHandler req resp = req -> IO (HandlerResponse resp)
194+type NewMessageHandler req resp = req -> IO resp
131195
132−serve :: forall req resp. (Show req, Serialise req, Serialise resp) =>
133− NewConnectionHandler resp ->
134− NewMessageHandler req resp ->
196+serve ::
197+ MessageHandlers ->
135198 HostAddr ->
136199 PortNumber ->
137200 Maybe (MVar SockAddr) ->
138201 IO Bool
139−serve connhandler userMsgHandler hostaddr port mSockLock = do
202+serve userMsgHandler hostaddr port mSockLock = do
140203 let
141− decoder :: NewMessageHandler req resp -> Decoder (Message req)
142− decoder _ =
143− case getDecoder (schema (Proxy @(Message req))) of
144− Left exc -> error (show exc)
145− Right dec -> dec
146204 handleSock sock = do
147− mResp <- connhandler
148205 lockingSocket <- newLock sock
149− case mResp of
150− Nothing -> pure ()
151− Just resp -> do
152− requestID <- UUID <$> UUIDBase.nextRandom
153− sendMessage (AsyncRequest requestID resp) lockingSocket
154− drainSocketMessages sock (decoder userMsgHandler) (serverMessageHandler lockingSocket userMsgHandler)
206+ drainSocketMessages sock (serverEnvelopeHandler lockingSocket userMsgHandler)
207+
155208 serially (S.unfold (SA.acceptOnAddrWith [(ReuseAddr,1)] mSockLock) (hostaddr, port)) & parallely . S.mapM (handleWithM handleSock) & S.drain
156209 pure True
157210
158−serverMessageHandler :: forall req resp. (Show req,
159− Serialise req,
160− Serialise resp)
161− => Locking Socket
162− -> NewMessageHandler req resp
163− -> Message req
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+
218+openEnvelope :: forall s. (Serialise s, Typeable s) => Envelope -> Maybe s
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-}
223+ if eprint == fingerprint (undefined :: s) then
224+ case deserialise bytes of
225+ Left err -> error (show err)
226+ Right v -> Just v
227+ --Just (evalDecoder decoder bytes)
228+ else
229+ Nothing
230+
231+serverEnvelopeHandler ::
232+ Locking Socket
233+ -> MessageHandlers
234+ -> Envelope
164235 -> IO ()
165−serverMessageHandler sockLock requestMessageHandler msg = do
236+serverEnvelopeHandler sockLock (MessageHandlers msgHandlers) envelope@(Envelope _ RequestMessage msgId _) = do
237+ --find first matching handler
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+{-
253+
254+
166255 --putStrLn $ "GOT MSG " ++ show msg
167256 let --runTimeout :: IO (HandlerResponse resp) -> IO (Maybe (HandlerResponse resp))
168257 runTimeout mTimeout m = case mTimeout of
@@ -190,32 +279,29 @@ serverMessageHandler sockLock requestMessageHandler msg = do
190279 --no response necessary
191280 ExceptionResponse{} -> putStrLn "client sent exception response"
192281 TimedOutResponse{} -> putStrLn "client sent timed out response"
282+-}
193283
194284--add callback to allow for responses via socket
195−type MessageHandler a = Message a -> IO ()
196−
197−type AsyncMessageHandler a = a -> IO ()
285+--type MessageHandler a = Message a -> IO ()
198286
287+type EnvelopeHandler = Envelope -> IO ()
199288
289+type AsyncMessageHandler a = a -> IO ()
200290
201−drainSocketMessages :: Serialise msg => Socket -> Decoder (Message msg) -> MessageHandler msg -> IO ()
202−drainSocketMessages sock decoder msgHandler = do
291+drainSocketMessages :: Socket -> EnvelopeHandler -> IO ()
292+drainSocketMessages sock envelopeHandler = do
203293 let sockStream = S.unfold readWithBufferOf (1024 * 4, sock)
204− handler bs = do
205− let decoded = evalDecoder decoder bs
206− msgHandler decoded
207− S.drain $ serially $ S.parseMany messageBoundaryP sockStream & S.mapM handler
294+ S.drain $ serially $ S.parseMany envelopeP sockStream & S.mapM envelopeHandler
208295
209296--send length-tagged bytestring, perhaps should be in network byte order?
210−sendMessage :: Serialise a => Message a -> Locking Socket -> IO ()
211−sendMessage msg sockLock = do
212− let
213− msgbytes = serialiseOnly msg
214− fullbytes = lenbytes <> msgbytes
215− len = BS.length msgbytes
216− lenbytes = BO.bytestring32 (fromIntegral len)
217−
297+sendEnvelope :: Envelope -> Locking Socket -> IO ()
298+sendEnvelope envelope sockLock = do
299+ let envelopebytes = encodeEnvelope envelope
300+ fullbytes = envelopebytes
218301 --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
219302 withLock sockLock $ \socket' ->
220303 Socket.sendAll socket' fullbytes
221− traceBytes "sendMessage" fullbytes
304+ traceBytes "sendEnvelope" fullbytes
305+
306+fingerprint :: Typeable a => a -> Fingerprint
307+fingerprint = typeRepFingerprint . typeOf