@@ -1,32 +1,41 @@
1−{-# LANGUAGE StandaloneDeriving, DerivingVia, DeriveGeneric, RankNTypes, TypeApplications, ScopedTypeVariables, MultiParamTypeClasses #-}
1+{-# LANGUAGE DerivingVia, DeriveGeneric, RankNTypes, ScopedTypeVariables, MultiParamTypeClasses, OverloadedStrings, GeneralizedNewtypeDeriving #-}
2+{-# HLINT ignore "Use lambda-case" #-}
23module Network.RPC.Curryer.Server where
34import Streamly
45import qualified Streamly.Prelude as S
5−import qualified Streamly.Network.Inet.TCP as TCP
66import Streamly.Network.Socket
77import Streamly.Internal.Network.Socket (handleWithM, writeChunk)
88import Network.Socket as Socket
99import Streamly.Internal.Data.Parser.ParserD as PD
1010import Codec.Winery
11+import Codec.Winery.Internal (varInt, decodeVarInt, getBytes)
12+import Codec.Winery.Class (mkExtractor)
1113import GHC.Generics
12−import Control.Concurrent.MVar (putMVar, MVar)
14+import Control.Concurrent.MVar (MVar)
1315import Control.Exception
1416import Data.Function ((&))
1517import Data.Word
16−import Control.Monad.Catch
1718import qualified Data.ByteString as BS
19+import qualified Data.ByteString.Lazy as BSL
20+import qualified Data.ByteString.FastBuilder as BB
1821import Streamly.Data.Fold as FL
1922import qualified Streamly.Internal.Data.Stream.IsStream as S
2023import Data.Foldable
21−import Data.Word
2224import Data.Bits
25+import qualified Data.Binary as B
26+import qualified Data.UUID as UUIDBase
2327import qualified Streamly.Internal.Data.Array.Storable.Foreign.Types as SA
28+import qualified Network.RPC.Curryer.StreamlyAdditions as SA
29+import Control.Monad
30+import Data.Hashable
2431
2532-- for toArrayS conversion
2633import qualified Data.ByteString.Internal as BSI
2734import Foreign.ForeignPtr (plusForeignPtr)
2835import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
29−import GHC.Ptr (minusPtr, plusPtr)
36+import GHC.Ptr (plusPtr)
37+
38+import Debug.Trace
3039
3140data ClientHelloMessage = ClientHelloMessage Int Int
3241 deriving Generic
@@ -42,59 +51,119 @@ data ServerHelloMessage = ServerHelloMessage { serverServices :: [String],
4251--client-side timeout
4352--server-to-client and client-to-server async request (no response requested)
4453
45−data Response a = Response a
54+data Message a = Response UUID a
55+ | AsyncRequest a
56+ | ResponseExpectedRequest UUID a
57+ -- | Services
4658 | ExceptionResponse String
47− deriving Generic
48− deriving Serialise via WineryVariant (Response a)
49−
50−data Request a = Request a
51− | Services
52− deriving Generic
53− deriving Serialise via WineryVariant (Request a)
54−
55−class Event a where
56− dispatchAsyncEvent :: a -> IO ()
57− dispatchEvent :: a -> IO (Response b)
59+ deriving (Generic, Show)
60+ deriving Serialise via WineryVariant (Message a)
61+
62+-- | The response type
63+data HandlerResponse a = NoResponse
64+ | HandlerResponse a
65+ | HandlerException a
66+ deriving Generic
67+ deriving Serialise via WineryVariant (HandlerResponse a)
68+
69+--avoid orphan instance
70+newtype UUID = UUID { _unUUID :: UUIDBase.UUID }
71+ deriving (Show, Eq, B.Binary, Hashable)
72+
73+instance Serialise UUID where
74+ schemaGen _ = pure (STag (TagStr "Data.UUID") SBytes)
75+ toBuilder uuid = let bytes = BSL.toStrict (B.encode uuid) in
76+ varInt (BS.length bytes) <> BB.byteString bytes
77+ {-# INLINE toBuilder #-}
78+ extractor = mkExtractor $
79+ \schema' -> case schema' of
80+ STag (TagStr "Data.UUID") SBytes ->
81+ pure $ \term -> case term of
82+ TBytes bs -> B.decode (BSL.fromStrict bs)
83+ term' -> throw (InvalidTerm term')
84+ x -> error $ "invalid schema element " <> show x
85+ decodeCurrent = B.decode . BSL.fromStrict <$> (decodeVarInt >>= getBytes)
86+
87+
88+data ConnectionError = CodecError String -- show of WineryException from exception initiator which cannot otherwise be transmitted over a line due to dependencies on TypeReps
89+ | TimeoutError
90+ deriving (Generic, Show, Eq)
91+ deriving Serialise via WineryVariant ConnectionError
5892
5993type HostAddr = (Word8, Word8, Word8, Word8)
6094
95+allHostAddrs,localHostAddr :: HostAddr
6196allHostAddrs = (0,0,0,0)
6297localHostAddr = (127,0,0,1)
6398
6499-- Each message is length-prefixed by a 32-bit unsigned length.
65100messageBoundaryP :: Parser IO Word8 BS.ByteString
66101messageBoundaryP = do
102+ --traceShowM "message!"
67103 w4x8 <- PD.take 4 FL.toList
104+ --traceShowM ("w4x8",w4x8)
68105 let c = fromIntegral (fromOctets w4x8)
106+ --traceShowM ("c", c)
69107 vals <- PD.take c FL.toList
70− pure (BS.pack vals)
108+ let bytes = BS.pack vals
109+ traceShowM ("vals", BS.length bytes, bytes)
110+ pure bytes
71111
72−serve :: Serialise msg =>
73− (msg -> Socket -> IO ()) ->
112+type NewConnectionHandler msg = IO (Maybe msg)
113+
114+type NewMessageHandler msg resp = msg -> IO (HandlerResponse resp)
115+
116+serve :: (Show msg, Serialise msg, Serialise resp) =>
117+ NewConnectionHandler resp ->
118+ NewMessageHandler msg resp ->
74119 HostAddr ->
75120 PortNumber ->
76− Maybe (MVar AddrInfo) ->
121+ Maybe (MVar SockAddr) ->
77122 IO Bool
78−serve handler hostaddr port mAddressMVar = do
123+serve connhandler msghandler hostaddr port mSockLock = do
79124 let handleSock sock = do
80− let sockStream :: SerialT IO Word8
81− sockStream = S.unfold readWithBufferOf (1024 * 4, sock)
82− messageHandler bs = case deserialise bs of
83− Left err -> error (show err) --FIXME
84− Right val -> handler val sock
125+ putStrLn "handleSock"
126+ let messageHandler msg = do
127+ putStrLn $ "GOT MSG " ++ show msg
128+ case msg of
129+ Response{} -> putStrLn "client sent response"
130+ ResponseExpectedRequest requestID val -> do
131+ putStrLn "ResponseExp"
132+ resp <- msghandler val
133+ case resp of
134+ HandlerResponse responseVal -> sendMessage (Response requestID responseVal) sock
135+ NoResponse -> error "attempt to return non-response to expected response message"
136+ HandlerException _ -> error "TODO HandlerException"
137+ AsyncRequest val -> do
138+ putStrLn "AsyncReq"
139+ void $ msghandler val
140+ --no response necessaryxs
141+ ExceptionResponse{} -> putStrLn "client sent exception response"
142+ -- allow the server to send an async welcome message to the new client, if necessary
143+ mResp <- connhandler
144+ case mResp of
145+ Nothing -> pure ()
146+ Just resp -> sendMessage (AsyncRequest resp) sock
85147 drainSocketMessages sock messageHandler
86− serially (S.unfold TCP.acceptOnAddr (hostaddr, port)) & parallely . S.mapM (handleWithM handleSock) & S.drain
148+ serially (S.unfold (SA.acceptOnAddrWith [(ReuseAddr,1)] mSockLock) (hostaddr, port)) & parallely . S.mapM (handleWithM handleSock) & S.drain
87149 pure True
88150
89−type MessageHandler a = a -> IO ()
151+--add callback to allow for responses via socket
152+type MessageHandler a = Message a -> IO ()
153+
154+type AsyncMessageHandler a = a -> IO ()
90155
91156drainSocketMessages :: Serialise msg => Socket -> MessageHandler msg -> IO ()
92157drainSocketMessages sock msgHandler = do
93158 let sockStream = S.unfold readWithBufferOf (1024 * 4, sock)
94− handler bs =
159+ handler bs = do
160+ --print ("drain handler", bs)
95161 case deserialise bs of
96− Left err -> error (show err) --FIXME
97− Right val -> msgHandler val --add response function
162+ Left err ->
163+ print err
164+ Right val -> do
165+ --putStrLn "deserialized value for msgHandler"
166+ msgHandler val --add response function
98167 S.drain $ S.parseManyD messageBoundaryP sockStream & S.mapM handler
99168
100169fromOctets :: [Word8] -> Word32
@@ -110,8 +179,16 @@ octets w =
110179 , fromIntegral w
111180 ]
112181
182+--send length-tagged bytestring, perhaps should be in network byte order?
113183sendMessage :: Serialise a => a -> Socket -> IO ()
114−sendMessage msg socket = writeChunk socket (toArrayS (serialise msg))
184+sendMessage msg socket' = do
185+ putStrLn ("sending bytes: " ++ show (blen <> bytes))
186+ writeChunk socket' (toArrayS (blen <> bytes))
187+ where
188+ bytes = serialise msg
189+ len = BS.length bytes
190+ blen = BS.pack (octets (fromIntegral len))
191+
115192
116193--from streamly-bytestring
117194-- | Convert a 'ByteString' to an array of 'Word8'. This function unwraps the
@@ -124,5 +201,5 @@ toArrayS :: BS.ByteString -> SA.Array Word8
124201toArrayS (BSI.PS fp off len) = SA.Array nfp endPtr
125202 where
126203 nfp = fp `plusForeignPtr` off
127− endPtr = unsafeForeignPtrToPtr nfp `plusPtr` len
204+ endPtr = unsafeForeignPtrToPtr nfp `plusPtr` len `plusPtr` 1 -- ? why +1?
128205