@@ -6,6 +6,7 @@ import qualified Streamly.Prelude as S
66import Streamly.Network.Socket
77import Streamly.Internal.Network.Socket (handleWithM, writeChunk)
88import Network.Socket as Socket
9+import Network.Socket.ByteString as Socket
910import Streamly.Internal.Data.Parser.ParserD as PD
1011import Codec.Winery
1112import Codec.Winery.Internal (varInt, decodeVarInt, getBytes)
@@ -100,14 +101,14 @@ localHostAddr = (127,0,0,1)
100101-- Each message is length-prefixed by a 32-bit unsigned length.
101102messageBoundaryP :: Parser IO Word8 BS.ByteString
102103messageBoundaryP = do
103− --traceShowM "message!"
104− w4x8 <- PD.take 4 FL.toList
105− --traceShowM ("w4x8",w4x8)
104+ let x = FL.toList
105+ w4x8 <- PD.take 4 x
106+ traceShowM ("w4x8", w4x8)
106107 let c = fromIntegral (fromOctets w4x8)
107− --traceShowM ("c", c)
108− vals <- PD.take c FL.toList
108+ traceShowM ("c", c)
109+ vals <- PD.take c x
109110 let bytes = BS.pack vals
110− traceShowM ("vals", BS.length bytes, bytes)
111+ traceShowM ("parsedBytes", c, BS.length bytes, bytes)
111112 pure bytes
112113
113114type NewConnectionHandler msg = IO (Maybe msg)
@@ -129,7 +130,7 @@ serve connhandler msghandler hostaddr port mSockLock = do
129130 case msg of
130131 Response{} -> putStrLn "client sent response"
131132 ResponseExpectedRequest requestID val -> do
132− putStrLn "ResponseExp"
133+ putStrLn "server received ResponseExpectedRequest"
133134 resp <- msghandler val
134135 case resp of
135136 HandlerResponse responseVal -> sendMessage (Response requestID responseVal) sock
@@ -148,7 +149,7 @@ serve connhandler msghandler hostaddr port mSockLock = do
148149 requestID <- UUID <$> UUIDBase.nextRandom
149150 sendMessage (AsyncRequest requestID resp) sock
150151 drainSocketMessages sock messageHandler
151− serially (S.unfold (SA.acceptOnAddrWith [(ReuseAddr,1)] mSockLock) (hostaddr, port)) & parallely . S.mapM (handleWithM handleSock) & S.drain
152+ serially (S.unfold (SA.acceptOnAddrWith [(ReuseAddr,1)] mSockLock) (hostaddr, port)) & serially . S.mapM (handleWithM handleSock) & S.drain
152153 pure True
153154
154155--add callback to allow for responses via socket
@@ -160,14 +161,12 @@ drainSocketMessages :: Serialise msg => Socket -> MessageHandler msg -> IO ()
160161drainSocketMessages sock msgHandler = do
161162 let sockStream = S.unfold readWithBufferOf (1024 * 4, sock)
162163 handler bs = do
163− --print ("drain handler", bs)
164164 case deserialise bs of
165165 Left err ->
166166 print err
167− Right val -> do
168− --putStrLn "deserialized value for msgHandler"
167+ Right val ->
169168 msgHandler val --add response function
170− S.drain $ S.parseManyD messageBoundaryP sockStream & S.mapM handler
169+ S.drain $ serially $ S.parseManyD messageBoundaryP sockStream & S.mapM handler
171170
172171fromOctets :: [Word8] -> Word32
173172fromOctets = foldl' accum 0
@@ -185,12 +184,18 @@ octets w =
185184--send length-tagged bytestring, perhaps should be in network byte order?
186185sendMessage :: Serialise a => a -> Socket -> IO ()
187186sendMessage msg socket' = do
188− putStrLn ("sending bytes: " ++ show (blen <> bytes))
189− writeChunk socket' (toArrayS (blen <> bytes))
190− where
191− bytes = serialise msg
192− len = BS.length bytes
193− blen = BS.pack (octets (fromIntegral len))
187+ let byteArray = toArraySlow fullbytes
188+ dbgByteString = BS.pack (SA.toList byteArray)
189+ msgbytes = serialise msg
190+ fullbytes = lenbytes <> msgbytes
191+ len = BS.length msgbytes
192+ lenbytes = BS.pack (octets (fromIntegral len))
193+ --when (dbgByteString /= msgbytes) (error "mismatch!")
194+
195+ byteCount <- Socket.send socket' fullbytes
196+ when (byteCount /= BS.length fullbytes) (error "bytes sent mismatch")
197+ traceShowM ("sent bytes:", byteCount, fullbytes)
198+
194199
195200
196201--from streamly-bytestring
@@ -206,3 +211,7 @@ toArrayS (BSI.PS fp off len) = SA.Array nfp endPtr
206211 nfp = fp `plusForeignPtr` off
207212 endPtr = unsafeForeignPtrToPtr nfp `plusPtr` len `plusPtr` 1 -- ? why +1?
208213
214+toArraySlow :: BS.ByteString -> SA.Array Word8
215+toArraySlow bs = traceShow ("arr", SA.length arr) arr
216+ where
217+ arr = SA.fromList (BS.unpack bs)