dmq: NtC applications

Author: crocodile-dentist

decentralized-message-queue/app/Main.hs

@@ -53,6 +53,10 @@ runDMQ cliopts@CLIOptions {
                              (decodeRemoteAddress (mapNtNDMQtoOuroboros maxBound)))
                   dmqLimitsAndTimeouts
                   defaultSigDecisionPolicy
+        dmqNtCApps =
+          ntcApps nodeKernel
+                  undefined -- codecs
+
         dmqDiffusionArguments =
           diffusionArguments debugTracer
                              debugTracer

decentralized-message-queue/decentralized-message-queue.cabal

@@ -54,6 +54,8 @@ library
     DMQ.Diffusion.PeerSelection
     DMQ.NodeToClient
     DMQ.NodeToNode
+    DMQ.NtC_Applications.LocalMsgSubmission
+    DMQ.NtC_Applications.LocalMsgNotification
     DMQ.Protocol.LocalMsgNotification.Client
     DMQ.Protocol.LocalMsgNotification.Codec
     DMQ.Protocol.LocalMsgNotification.Server

decentralized-message-queue/src/DMQ/NodeToClient.hs

@@ -1,3 +1,4 @@
+{-# LANGUAGE OverloadedRecordDot #-}
 {-# LANGUAGE DeriveAnyClass #-}
 {-# LANGUAGE DeriveGeneric  #-}
 {-# LANGUAGE LambdaCase     #-}
@@ -9,25 +10,41 @@ import Codec.CBOR.Term qualified as CBOR
 import Control.DeepSeq (NFData)
 import Control.Monad ((>=>))
 import Data.Bits (Bits (..))
+import Data.ByteString.Lazy (ByteString)
 import Data.Text (Text)
 import Data.Text qualified as T
 import GHC.Generics (Generic)
 
+import Control.Monad.Class.MonadFork
 import Control.Monad.Class.MonadST (MonadST)
+import Control.Monad.Class.MonadThrow
 import Control.Tracer (Tracer, nullTracer)
 
 import Network.Mux qualified as Mx
 
 import Ouroboros.Network.CodecCBORTerm (CodecCBORTerm (..))
 import Ouroboros.Network.ConnectionId (ConnectionId)
-import Ouroboros.Network.Driver.Simple (TraceSendRecv)
+import Ouroboros.Network.Driver.Simple
 import Ouroboros.Network.Handshake.Acceptable (Acceptable (..))
 import Ouroboros.Network.Handshake.Queryable (Queryable (..))
 import Ouroboros.Network.Magic (NetworkMagic (..))
 import Ouroboros.Network.Protocol.Handshake (Accept (..), Handshake,
            HandshakeArguments (..))
 import Ouroboros.Network.Protocol.Handshake.Codec (cborTermVersionDataCodec,
            codecHandshake, noTimeLimitsHandshake)
+import DMQ.Diffusion.NodeKernel
+import Network.TypedProtocol.Codec
+import DMQ.Protocol.LocalMsgSubmission.Type
+import qualified Network.TypedProtocol.Codec.CBOR as CBOR
+import DMQ.Protocol.LocalMsgNotification.Type
+import DMQ.Protocol.LocalMsgNotification.Server
+import DMQ.Protocol.LocalMsgSubmission.Server
+import Ouroboros.Network.Context
+import DMQ.Protocol.SigSubmission.Type
+import qualified Ouroboros.Network.TxSubmission.Mempool.Simple as Mempool
+import Control.Concurrent.Class.MonadSTM
+import DMQ.NtC_Applications.LocalMsgSubmission
+import DMQ.NtC_Applications.LocalMsgNotification
 
 data NodeToClientVersion =
   NodeToClientV_1
@@ -106,6 +123,7 @@ nodeToClientCodecCBORTerm _v = CodecCBORTerm {encodeTerm, decodeTerm}
       decoder x query | x >= 0 && x <= 0xffffffff = Right (NodeToClientVersionData (NetworkMagic $ fromIntegral x) query)
                       | otherwise                 = Left $ T.pack $ "networkMagic out of bound: " <> show x
 
+
 data Protocols =
   Protocols {
   }
@@ -139,3 +157,74 @@ stdVersionDataNTC networkMagic =
     { networkMagic
     , query        = False
     }
+
+type LocalMsgSubmission' = LocalMsgSubmission Sig Int
+type LocalMsgNotification' = LocalMsgNotification Sig
+
+data Codecs m =
+  Codecs {
+    msgSubmissionCodec
+      :: Codec LocalMsgSubmission'
+               CBOR.DeserialiseFailure m ByteString
+  , msgNotificationCodec
+      :: Codec LocalMsgNotification'
+               CBOR.DeserialiseFailure m ByteString
+  }
+
+
+-- | A node-to-client application
+--
+type App ctx m a =
+     ctx
+  -> Mx.Channel m ByteString
+  -> m (a, Maybe ByteString)
+
+
+data Apps addr m a =
+  Apps {
+    -- | Start a sig-submission client
+    aLocalMsgSubmission   :: App addr m a
+
+    -- | Start a sig-submission server
+  , aLocalMsgNotification :: App addr m a
+  }
+
+
+-- | Construct applications for the node-to-client protocols
+--
+ntcApps
+  :: (MonadThrow m, MonadThread m, MonadSTM m)
+  => NodeKernel addr m
+  -> Codecs m
+  -> Apps addr m ()
+ntcApps NodeKernel { mempool } codecs =
+  Apps {
+    aLocalMsgSubmission
+  , aLocalMsgNotification
+  }
+  where
+    sigSize :: Sig -> SizeInBytes
+    sigSize _ = 0 -- TODO
+
+    mempoolReader = Mempool.getReader sigId sigSize mempool
+    mempoolWriter = Mempool.getWriter sigId (const True) mempool
+
+    -- aLocalMsgSubmission :: _
+    aLocalMsgSubmission _ctx channel = do
+      labelThisThread "LocalMsgSubmissionServer"
+      runPeer
+        undefined
+        codecs.msgSubmissionCodec
+        channel
+        (localMsgSubmissionServerPeer $
+          localMsgSubmissionServer undefined mempoolWriter)
+
+    -- aLocalMsgNotification :: _
+    aLocalMsgNotification _ctx channel = do
+      labelThisThread "LocalMsgNotificationServer"
+      runPeer
+        undefined
+        codecs.msgNotificationCodec
+        channel
+        (localMsgNotificationServerPeer $
+          localMsgNotificationServer undefined)

decentralized-message-queue/src/DMQ/NtC_Applications/LocalMsgNotification.hs

@@ -0,0 +1,17 @@
+module DMQ.NtC_Applications.LocalMsgNotification where
+
+import DMQ.Protocol.LocalMsgNotification.Server
+import Control.Concurrent.Class.MonadSTM
+
+localMsgNotificationServer
+  :: (MonadSTM m)
+  => a
+  -> LocalMsgNotificationServer m msg b
+localMsgNotificationServer _a =
+  LocalMsgNotificationServer . pure $ serverIdle undefined
+  where
+    serverIdle :: b -> ServerIdle m msg b
+    serverIdle _b = ServerIdle { msgRequestHandler, msgDoneHandler }
+
+    msgRequestHandler = undefined
+    msgDoneHandler = undefined

decentralized-message-queue/src/DMQ/NtC_Applications/LocalMsgSubmission.hs

@@ -0,0 +1,38 @@
+module DMQ.NtC_Applications.LocalMsgSubmission where
+
+import Control.Tracer
+import Control.Concurrent.Class.MonadSTM
+import Ouroboros.Network.TxSubmission.Mempool.Simple
+import DMQ.Protocol.LocalMsgSubmission.Server
+import Ouroboros.Network.TxSubmission.Inbound.V2
+import Data.Maybe
+import DMQ.Protocol.LocalMsgSubmission.Type
+
+-- | Local transaction submission server, for adding txs to the 'Mempool'
+--
+localMsgSubmissionServer ::
+     MonadSTM m
+  => Tracer m TraceLocalMsgSubmission
+  -> TxSubmissionMempoolWriter msgid msg idx m
+  -> m (LocalMsgSubmissionServer msg reject m ())
+localMsgSubmissionServer tracer TxSubmissionMempoolWriter { mempoolAddTxs } =
+    pure server
+  where
+    failure = (SubmitFail undefined, server)
+    success = const (SubmitSuccess, server)
+
+    server = LocalTxSubmissionServer {
+      recvMsgSubmitTx = \msg -> do
+        traceWith tracer $ TraceReceivedMsg {- TODO: add msg argument -}
+        maybe failure success . listToMaybe <$> mempoolAddTxs [msg]
+        -- case addTxRes of
+        --   MempoolTxAdded _tx             -> return (SubmitSuccess, server)
+        --   MempoolTxRejected _tx addTxErr -> return (SubmitFail addTxErr, server)
+
+    , recvMsgDone = ()
+    }
+
+
+data TraceLocalMsgSubmission
+  = TraceReceivedMsg
+    -- ^ A transaction was received.