@@ -0,0 +1,128 @@
1+{-# LANGUAGE StandaloneDeriving, DerivingVia, DeriveGeneric, RankNTypes, TypeApplications, ScopedTypeVariables, MultiParamTypeClasses #-}
2+module Network.RPC.Vintner.Server where
3+import Streamly
4+import qualified Streamly.Prelude as S
5+import qualified Streamly.Network.Inet.TCP as TCP
6+import Streamly.Network.Socket
7+import Streamly.Internal.Network.Socket (handleWithM, writeChunk)
8+import Network.Socket as Socket
9+import Streamly.Internal.Data.Parser.ParserD as PD
10+import Codec.Winery
11+import GHC.Generics
12+import Control.Concurrent.MVar (putMVar, MVar)
13+import Control.Exception
14+import Data.Function ((&))
15+import Data.Word
16+import Control.Monad.Catch
17+import qualified Data.ByteString as BS
18+import Streamly.Data.Fold as FL
19+import qualified Streamly.Internal.Data.Stream.IsStream as S
20+import Data.Foldable
21+import Data.Word
22+import Data.Bits
23+import qualified Streamly.Internal.Data.Array.Storable.Foreign.Types as SA
24+
25+-- for toArrayS conversion
26+import qualified Data.ByteString.Internal as BSI
27+import Foreign.ForeignPtr (plusForeignPtr)
28+import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
29+import GHC.Ptr (minusPtr, plusPtr)
30+
31+data ClientHelloMessage = ClientHelloMessage Int Int
32+ deriving Generic
33+ deriving Serialise via WineryVariant ClientHelloMessage
34+
35+data ServerHelloMessage = ServerHelloMessage { serverServices :: [String],
36+ serverInfo :: String
37+ }
38+ deriving Generic
39+ deriving Serialise via WineryRecord ServerHelloMessage
40+
41+--request-response token
42+--client-side timeout
43+--server-to-client and client-to-server async request (no response requested)
44+
45+data Response a = Response a
46+ | 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)
58+
59+type HostAddr = (Word8, Word8, Word8, Word8)
60+
61+allHostAddrs = (0,0,0,0)
62+localHostAddr = (127,0,0,1)
63+
64+-- Each message is length-prefixed by a 32-bit unsigned length.
65+messageBoundaryP :: Parser IO Word8 BS.ByteString
66+messageBoundaryP = do
67+ w4x8 <- PD.take 4 FL.toList
68+ let c = fromIntegral (fromOctets w4x8)
69+ vals <- PD.take c FL.toList
70+ pure (BS.pack vals)
71+
72+serve :: Serialise msg =>
73+ (msg -> Socket -> IO ()) ->
74+ HostAddr ->
75+ PortNumber ->
76+ Maybe (MVar AddrInfo) ->
77+ IO Bool
78+serve handler hostaddr port mAddressMVar = do
79+ 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
85+ drainSocketMessages sock messageHandler
86+ serially (S.unfold TCP.acceptOnAddr (hostaddr, port)) & parallely . S.mapM (handleWithM handleSock) & S.drain
87+ pure True
88+
89+type MessageHandler a = a -> IO ()
90+
91+drainSocketMessages :: Serialise msg => Socket -> MessageHandler msg -> IO ()
92+drainSocketMessages sock msgHandler = do
93+ let sockStream = S.unfold readWithBufferOf (1024 * 4, sock)
94+ handler bs =
95+ case deserialise bs of
96+ Left err -> error (show err) --FIXME
97+ Right val -> msgHandler val --add response function
98+ S.drain $ S.parseManyD messageBoundaryP sockStream & S.mapM handler
99+
100+fromOctets :: [Word8] -> Word32
101+fromOctets = foldl' accum 0
102+ where
103+ accum a o = (a `shiftL` 8) .|. fromIntegral o
104+
105+octets :: Word32 -> [Word8]
106+octets w =
107+ [ fromIntegral (w `shiftR` 24)
108+ , fromIntegral (w `shiftR` 16)
109+ , fromIntegral (w `shiftR` 8)
110+ , fromIntegral w
111+ ]
112+
113+sendMessage :: Serialise a => a -> Socket -> IO ()
114+sendMessage msg socket = writeChunk socket (toArrayS (serialise msg))
115+
116+--from streamly-bytestring
117+-- | Convert a 'ByteString' to an array of 'Word8'. This function unwraps the
118+-- 'ByteString' and wraps it with 'Array' constructors and hence the operation
119+-- is performed in constant time.
120+{-# INLINE toArrayS #-}
121+toArrayS :: BS.ByteString -> SA.Array Word8
122+--slow path
123+--toArrayS = SA.fromList . BS.unpack
124+toArrayS (BSI.PS fp off len) = SA.Array nfp endPtr
125+ where
126+ nfp = fp `plusForeignPtr` off
127+ endPtr = unsafeForeignPtrToPtr nfp `plusPtr` len
128+