dmq: NtC applications

Author: crocodile-dentist

decentralized-message-queue/app/Main.hs

@@ -17,14 +17,19 @@ import DMQ.Configuration.CLIOptions (parseCLIOptions)
 import DMQ.Configuration.Topology (readTopologyFileOrError)
 import DMQ.Diffusion.Applications (diffusionApplications)
 import DMQ.Diffusion.Arguments
-import DMQ.Diffusion.NodeKernel (withNodeKernel)
+import DMQ.Diffusion.NodeKernel (mempool, withNodeKernel)
+import DMQ.NodeToClient qualified as NtC
 import DMQ.NodeToNode (dmqCodecs, dmqLimitsAndTimeouts, ntnApps)
+import DMQ.Protocol.LocalMsgSubmission.Codec
+import DMQ.Protocol.SigSubmission.Codec
+import DMQ.Protocol.SigSubmission.Type (Sig (..))
 import DMQ.Tracer
 
 import DMQ.Diffusion.PeerSelection (policy)
 import Ouroboros.Network.Diffusion qualified as Diffusion
 import Ouroboros.Network.PeerSelection.PeerSharing.Codec (decodeRemoteAddress,
            encodeRemoteAddress)
+import Ouroboros.Network.TxSubmission.Mempool.Simple qualified as Mempool
 
 main :: IO ()
 main = void . runDMQ =<< execParser opts
@@ -78,6 +83,13 @@ runDMQ commandLineConfig = do
                                (decodeRemoteAddress maxBound))
                     dmqLimitsAndTimeouts
                     defaultSigDecisionPolicy
+          dmqNtCApps =
+            let sigSize _ = 0 -- TODO
+                maxMsgs = 1000 -- TODO: make this dynamic?
+                mempoolReader = Mempool.getReader sigId sigSize (mempool nodeKernel)
+                mempoolWriter = Mempool.getWriter sigId (const True) (mempool nodeKernel)
+             in NtC.ntcApps mempoolReader mempoolWriter maxMsgs
+                            (NtC.dmqCodecs encodeSig decodeSig encodeReject decodeReject)
           dmqDiffusionArguments =
             diffusionArguments (if handshakeTracer
                                   then WithEventType "Handshake" >$< tracer
@@ -91,6 +103,7 @@ runDMQ commandLineConfig = do
                                   dmqDiffusionConfiguration
                                   dmqLimitsAndTimeouts
                                   dmqNtNApps
+                                  dmqNtCApps
                                   (policy policyRng)
 
       Diffusion.run dmqDiffusionArguments

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

@@ -44,6 +44,7 @@ library
   import:
     warnings,
     extensions
+
   exposed-modules:
     DMQ.Configuration
     DMQ.Configuration.CLIOptions
@@ -53,11 +54,14 @@ library
     DMQ.Diffusion.NodeKernel
     DMQ.Diffusion.PeerSelection
     DMQ.NodeToClient
+    DMQ.NodeToClient.LocalMsgSubmission
+    DMQ.NodeToClient.LocalMsgNotification
     DMQ.NodeToClient.Version
     DMQ.NodeToNode
     DMQ.NodeToNode.Version
     DMQ.Protocol.LocalMsgNotification.Client
     DMQ.Protocol.LocalMsgNotification.Codec
+    DMQ.Protocol.LocalMsgNotification.Examples
     DMQ.Protocol.LocalMsgNotification.Server
     DMQ.Protocol.LocalMsgNotification.Type
     DMQ.Protocol.LocalMsgSubmission.Client
@@ -106,6 +110,7 @@ executable dmq-exe
   import:
     warnings,
     extensions
+
   main-is: Main.hs
   ghc-options:
     -threaded
@@ -126,15 +131,17 @@ executable dmq-exe
   default-language: Haskell2010
 
 test-suite dmq-test
-  import: warnings,
-          extensions
+  import:
+    warnings,
+    extensions
+
   default-language: Haskell2010
   other-modules:
     Test.DMQ.NodeToClient
     Test.DMQ.NodeToNode
-    Test.DMQ.Protocol.SigSubmission
     Test.DMQ.Protocol.LocalMsgNotification
     Test.DMQ.Protocol.LocalMsgSubmission
+    Test.DMQ.Protocol.SigSubmission
 
   type: exitcode-stdio-1.0
   hs-source-dirs: test
@@ -143,8 +150,12 @@ test-suite dmq-test
     QuickCheck,
     base >=4.14 && <4.22,
     bytestring,
+    contra-tracer,
     decentralized-message-queue,
+    io-classes,
+    io-sim,
     ouroboros-network-api,
+    ouroboros-network-framework,
     ouroboros-network-protocols:testlib,
     ouroboros-network-testing,
     quickcheck-instances,
@@ -154,7 +165,6 @@ test-suite dmq-test
     time,
     typed-protocols:{typed-protocols, codec-properties},
     with-utf8,
-    io-classes
 
   ghc-options:
     -fno-ignore-asserts

decentralized-message-queue/src/DMQ/Diffusion/Applications.hs

@@ -8,13 +8,13 @@ import DMQ.Configuration
 import DMQ.Diffusion.NodeKernel (NodeKernel (..))
 import DMQ.NodeToClient (NodeToClientVersion, NodeToClientVersionData,
            stdVersionDataNTC)
+import DMQ.NodeToClient qualified as NTC
 import DMQ.NodeToNode (NodeToNodeVersion, NodeToNodeVersionData,
            stdVersionDataNTN)
 import DMQ.NodeToNode qualified as NTN
 
 import Ouroboros.Network.Diffusion.Types qualified as Diffusion
 import Ouroboros.Network.ExitPolicy (RepromoteDelay (..))
-import Ouroboros.Network.Mux (OuroborosApplication (..))
 import Ouroboros.Network.PeerSelection.Governor.Types (PeerSelectionPolicy)
 import Ouroboros.Network.Protocol.Handshake.Version (combineVersions,
            simpleSingletonVersions)
@@ -27,6 +27,7 @@ diffusionApplications
   -> Diffusion.Configuration NoExtraFlags m ntnFd ntnAddr ntcFd ntcAddr
   -> NTN.LimitsAndTimeouts ntnAddr
   -> NTN.Apps ntnAddr m a ()
+  -> NTC.Apps ntcAddr m ()
   -> PeerSelectionPolicy ntnAddr m
   -> Diffusion.Applications ntnAddr NodeToNodeVersion   NodeToNodeVersionData
                             ntcAddr NodeToClientVersion NodeToClientVersionData
@@ -44,6 +45,7 @@ diffusionApplications
   }
   ntnLimitsAndTimeouts
   ntnApps
+  ntcApps
   peerSelectionPolicy =
   Diffusion.Applications {
     daApplicationInitiatorMode =
@@ -67,11 +69,7 @@ diffusionApplications
         [ simpleSingletonVersions
             version
             (stdVersionDataNTC networkMagic)
-            (\_versionData ->
-                OuroborosApplication
-                  [
-                  ]
-            )
+            (NTC.responders ntcApps version)
         | version <- [minBound..maxBound]
         ]
   , daRethrowPolicy                     =  muxErrorRethrowPolicy

decentralized-message-queue/src/DMQ/Diffusion/NodeKernel.hs

@@ -39,16 +39,16 @@ import DMQ.Protocol.SigSubmission.Type (Sig (..), SigId)
 data NodeKernel ntnAddr m =
   NodeKernel {
     -- | The fetch client registry, used for the keep alive clients.
-    fetchClientRegistry :: FetchClientRegistry (ConnectionId ntnAddr) () () m
+    fetchClientRegistry :: !(FetchClientRegistry (ConnectionId ntnAddr) () () m)
 
     -- | Read the current peer sharing registry, used for interacting with
     -- the PeerSharing protocol
-  , peerSharingRegistry :: PeerSharingRegistry ntnAddr m
-  , peerSharingAPI      :: PeerSharingAPI ntnAddr StdGen m
-  , mempool             :: Mempool m Sig
-  , sigChannelVar       :: TxChannelsVar m ntnAddr SigId Sig
-  , sigMempoolSem       :: TxMempoolSem m
-  , sigSharedTxStateVar :: SharedTxStateVar m ntnAddr SigId Sig
+  , peerSharingRegistry :: !(PeerSharingRegistry ntnAddr m)
+  , peerSharingAPI      :: !(PeerSharingAPI ntnAddr StdGen m)
+  , mempool             :: !(Mempool m Sig)
+  , sigChannelVar       :: !(TxChannelsVar m ntnAddr SigId Sig)
+  , sigMempoolSem       :: !(TxMempoolSem m)
+  , sigSharedTxStateVar :: !(SharedTxStateVar m ntnAddr SigId Sig)
   }
 
 newNodeKernel :: ( MonadLabelledSTM m