{-# LANGUAGE CPP                 #-}
{-# LANGUAGE DataKinds           #-}
{-# LANGUAGE DerivingVia         #-}
{-# LANGUAGE FlexibleContexts    #-}
{-# LANGUAGE GADTs               #-}
{-# LANGUAGE NamedFieldPuns      #-}
{-# LANGUAGE RankNTypes          #-}
{-# LANGUAGE ScopedTypeVariables #-}
{-# LANGUAGE TypeApplications    #-}
{-# LANGUAGE TypeOperators       #-}

-- it is useful to have 'HasInitiator' constraint on 'connectToNode' & friends.
{-# OPTIONS_GHC -Wno-redundant-constraints #-}

-- For Hashable SockAddr
{-# OPTIONS_GHC -Wno-orphans #-}


-- |
-- Module exports interface for running a node over a socket over TCP \/ IP.
--
module Ouroboros.Network.Socket
  ( -- * High level socket interface
    ConnectionTable
  , ConnectionTableRef (..)
  , ValencyCounter
  , SomeResponderApplication (..)
  , ConnectionId (..)
  , ConnectToArgs (..)
  , connectToNode
  , connectToNodeWithMux
  , connectToNodeSocket
  , connectToNode'
  , connectToNodeWithMux'
    -- * Socket configuration
  , configureSocket
  , configureSystemdSocket
  , SystemdSocketTracer (..)
    -- * Traces
  , NetworkConnectTracers (..)
  , nullNetworkConnectTracers
  , debuggingNetworkConnectTracers
    -- * Re-export of HandshakeCallbacks
  , HandshakeCallbacks (..)
    -- * Re-export connection table functions
  , newConnectionTable
  , refConnection
  , addConnection
  , removeConnection
  , newValencyCounter
  , addValencyCounter
  , remValencyCounter
  , waitValencyCounter
  , readValencyCounter
    -- * Auxiliary functions
  , sockAddrFamily
  , simpleMuxCallback
  ) where

import Codec.CBOR.Read qualified as CBOR
import Codec.CBOR.Term qualified as CBOR
import Control.Applicative (Alternative)
import Control.Concurrent.Class.MonadSTM.Strict
import Control.Monad (unless, when)
import Control.Monad.Class.MonadAsync
import Control.Monad.Class.MonadThrow
import Control.Monad.Class.MonadTime.SI
import Data.Bifunctor (first)
import Data.ByteString.Lazy qualified as BL
import Data.Foldable (traverse_)
import Data.Hashable
import Data.Monoid.Synchronisation (FirstToFinish (..))
import Data.Typeable (Typeable)
import Data.Word (Word16)

import Network.Socket (SockAddr, Socket, StructLinger (..))
import Network.Socket qualified as Socket

import Control.Tracer

import Network.Mux qualified as Mx
import Network.Mux.Bearer qualified as Mx
import Network.Mux.DeltaQ.TraceTransformer
import Network.TypedProtocol.Codec hiding (decode, encode)

import Ouroboros.Network.Context
import Ouroboros.Network.Driver.Limits
import Ouroboros.Network.Handshake (HandshakeCallbacks (..))
import Ouroboros.Network.IOManager (IOManager)
import Ouroboros.Network.Mux
import Ouroboros.Network.Protocol.Handshake
import Ouroboros.Network.Protocol.Handshake.Codec
import Ouroboros.Network.Protocol.Handshake.Type
import Ouroboros.Network.Server.ConnectionTable
import Ouroboros.Network.Snocket (Snocket)
import Ouroboros.Network.Snocket qualified as Snocket


-- | Tracer used by 'connectToNode' (and derivatives, like
-- 'Ouroboros.Network.NodeToNode.connectTo' or
-- 'Ouroboros.Network.NodeToClient.connectTo).
--
data NetworkConnectTracers addr vNumber = NetworkConnectTracers {
      forall addr vNumber.
NetworkConnectTracers addr vNumber
-> Tracer IO (WithBearer (ConnectionId addr) Trace)
nctMuxTracer         :: Tracer IO (Mx.WithBearer (ConnectionId addr)  Mx.Trace),
      -- ^ low level mux-network tracer, which logs mux sdu (send and received)
      -- and other low level multiplexing events.
      forall addr vNumber.
NetworkConnectTracers addr vNumber
-> Tracer
     IO
     (WithBearer
        (ConnectionId addr) (TraceSendRecv (Handshake vNumber Term)))
nctHandshakeTracer   :: Tracer IO (Mx.WithBearer (ConnectionId addr)
                                          (TraceSendRecv (Handshake vNumber CBOR.Term)))
      -- ^ handshake protocol tracer; it is important for analysing version
      -- negotiation mismatches.
    }

nullNetworkConnectTracers :: NetworkConnectTracers addr vNumber
nullNetworkConnectTracers :: forall addr vNumber. NetworkConnectTracers addr vNumber
nullNetworkConnectTracers = NetworkConnectTracers {
      nctMuxTracer :: Tracer IO (WithBearer (ConnectionId addr) Trace)
nctMuxTracer       = Tracer IO (WithBearer (ConnectionId addr) Trace)
forall (m :: * -> *) a. Applicative m => Tracer m a
nullTracer,
      nctHandshakeTracer :: Tracer
  IO
  (WithBearer
     (ConnectionId addr) (TraceSendRecv (Handshake vNumber Term)))
nctHandshakeTracer = Tracer
  IO
  (WithBearer
     (ConnectionId addr) (TraceSendRecv (Handshake vNumber Term)))
forall (m :: * -> *) a. Applicative m => Tracer m a
nullTracer
    }


debuggingNetworkConnectTracers :: (Show addr, Show vNumber)
                               => NetworkConnectTracers addr vNumber
debuggingNetworkConnectTracers :: forall addr vNumber.
(Show addr, Show vNumber) =>
NetworkConnectTracers addr vNumber
debuggingNetworkConnectTracers = NetworkConnectTracers {
      nctMuxTracer :: Tracer IO (WithBearer (ConnectionId addr) Trace)
nctMuxTracer       = Tracer IO String
-> Tracer IO (WithBearer (ConnectionId addr) Trace)
forall a (m :: * -> *). Show a => Tracer m String -> Tracer m a
showTracing Tracer IO String
forall (m :: * -> *). MonadIO m => Tracer m String
stdoutTracer,
      nctHandshakeTracer :: Tracer
  IO
  (WithBearer
     (ConnectionId addr) (TraceSendRecv (Handshake vNumber Term)))
nctHandshakeTracer = Tracer IO String
-> Tracer
     IO
     (WithBearer
        (ConnectionId addr) (TraceSendRecv (Handshake vNumber Term)))
forall a (m :: * -> *). Show a => Tracer m String -> Tracer m a
showTracing Tracer IO String
forall (m :: * -> *). MonadIO m => Tracer m String
stdoutTracer
    }

sockAddrFamily
    :: Socket.SockAddr
    -> Socket.Family
sockAddrFamily :: SockAddr -> Family
sockAddrFamily (Socket.SockAddrInet  PortNumber
_ HostAddress
_    ) = Family
Socket.AF_INET
sockAddrFamily (Socket.SockAddrInet6 PortNumber
_ HostAddress
_ HostAddress6
_ HostAddress
_) = Family
Socket.AF_INET6
sockAddrFamily (Socket.SockAddrUnix String
_       ) = Family
Socket.AF_UNIX


-- | Configure a socket.  Either 'Socket.AF_INET' or 'Socket.AF_INET6' socket
-- is expected.
--
configureSocket :: Socket -> Maybe SockAddr -> IO ()
configureSocket :: Socket -> Maybe SockAddr -> IO ()
configureSocket Socket
sock Maybe SockAddr
addr = do
    let fml :: Maybe Family
fml = SockAddr -> Family
sockAddrFamily (SockAddr -> Family) -> Maybe SockAddr -> Maybe Family
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Maybe SockAddr
addr
    Socket -> SocketOption -> Int -> IO ()
Socket.setSocketOption Socket
sock SocketOption
Socket.ReuseAddr Int
1
#if !defined(mingw32_HOST_OS)
    -- not supported on Windows 10
    Socket -> SocketOption -> Int -> IO ()
Socket.setSocketOption Socket
sock SocketOption
Socket.ReusePort Int
1
#endif
    Socket -> SocketOption -> Int -> IO ()
Socket.setSocketOption Socket
sock SocketOption
Socket.NoDelay Int
1
    -- it is safe to set 'SO_LINGER' option (which implicates that every
    -- close will reset the connection), since our protocols are robust.
    -- In particular if invalid data will arrive (which includes the rare
    -- case of a late packet from a previous connection), we will abandon
    -- (and close) the connection.
    Socket -> SocketOption -> StructLinger -> IO ()
forall a. Storable a => Socket -> SocketOption -> a -> IO ()
Socket.setSockOpt Socket
sock SocketOption
Socket.Linger
                          (StructLinger { sl_onoff :: CInt
sl_onoff  = CInt
1,
                                          sl_linger :: CInt
sl_linger = CInt
0 })
    Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (Maybe Family
fml Maybe Family -> Maybe Family -> Bool
forall a. Eq a => a -> a -> Bool
== Family -> Maybe Family
forall a. a -> Maybe a
Just Family
Socket.AF_INET6)
      -- An AF_INET6 socket can be used to talk to both IPv4 and IPv6 end points, and
      -- it is enabled by default on some systems. Disabled here since we run a separate
      -- IPv4 server instance if configured to use IPv4.
      (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ Socket -> SocketOption -> Int -> IO ()
Socket.setSocketOption Socket
sock SocketOption
Socket.IPv6Only Int
1


-- | Configure sockets passed through systemd socket activation.
-- Currently 'ReuseAddr' and 'Linger' options are not configurable with
-- 'systemd.socket', these options are set by this function.  For other socket
-- options we only trace if they are not set.
--
configureSystemdSocket :: Tracer IO SystemdSocketTracer -> Socket -> SockAddr -> IO ()
configureSystemdSocket :: Tracer IO SystemdSocketTracer -> Socket -> SockAddr -> IO ()
configureSystemdSocket Tracer IO SystemdSocketTracer
tracer Socket
sock SockAddr
addr = do
   let fml :: Family
fml = SockAddr -> Family
sockAddrFamily SockAddr
addr
   case Family
fml of
     Family
Socket.AF_INET ->
          Socket -> SocketOption -> Int -> IO ()
Socket.setSocketOption Socket
sock SocketOption
Socket.ReuseAddr Int
1
     Family
Socket.AF_INET6 ->
          Socket -> SocketOption -> Int -> IO ()
Socket.setSocketOption Socket
sock SocketOption
Socket.ReuseAddr Int
1
     Family
_ -> () -> IO ()
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return ()
#if !defined(mingw32_HOST_OS)
   -- not supported on Windows 10
   reusePortOpt <- Socket -> SocketOption -> IO Int
Socket.getSocketOption Socket
sock SocketOption
Socket.ReusePort
   unless (reusePortOpt /= 0) $
     traceWith tracer (SocketOptionNotSet Socket.ReusePort)
#endif
   noDelayOpt <- Socket.getSocketOption sock Socket.NoDelay
   unless (noDelayOpt /= 0) $
     traceWith tracer (SocketOptionNotSet Socket.NoDelay)

   Socket.setSockOpt sock Socket.Linger
                         (StructLinger { sl_onoff  = 1,
                                         sl_linger = 0 })
   when (fml == Socket.AF_INET6) $ do
     ipv6OnlyOpt <- Socket.getSocketOption sock Socket.IPv6Only
     unless (ipv6OnlyOpt /= 0) $
       traceWith tracer (SocketOptionNotSet Socket.IPv6Only)

data SystemdSocketTracer = SocketOptionNotSet Socket.SocketOption
  deriving Int -> SystemdSocketTracer -> ShowS
[SystemdSocketTracer] -> ShowS
SystemdSocketTracer -> String
(Int -> SystemdSocketTracer -> ShowS)
-> (SystemdSocketTracer -> String)
-> ([SystemdSocketTracer] -> ShowS)
-> Show SystemdSocketTracer
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> SystemdSocketTracer -> ShowS
showsPrec :: Int -> SystemdSocketTracer -> ShowS
$cshow :: SystemdSocketTracer -> String
show :: SystemdSocketTracer -> String
$cshowList :: [SystemdSocketTracer] -> ShowS
showList :: [SystemdSocketTracer] -> ShowS
Show


instance Hashable Socket.SockAddr where
  hashWithSalt :: Int -> SockAddr -> Int
hashWithSalt Int
s (Socket.SockAddrInet   PortNumber
p   HostAddress
a   ) = Int -> (Word16, HostAddress) -> Int
forall a. Hashable a => Int -> a -> Int
hashWithSalt Int
s (PortNumber -> Word16
forall a b. (Integral a, Num b) => a -> b
fromIntegral PortNumber
p :: Word16, HostAddress
a)
  hashWithSalt Int
s (Socket.SockAddrInet6  PortNumber
p HostAddress
_ HostAddress6
a HostAddress
_ ) = Int -> (Word16, HostAddress6) -> Int
forall a. Hashable a => Int -> a -> Int
hashWithSalt Int
s (PortNumber -> Word16
forall a b. (Integral a, Num b) => a -> b
fromIntegral PortNumber
p :: Word16, HostAddress6
a)
  hashWithSalt Int
s (Socket.SockAddrUnix   String
p       ) = Int -> String -> Int
forall a. Hashable a => Int -> a -> Int
hashWithSalt Int
s String
p

-- | We place an upper limit of `30s` on the time we wait on receiving an SDU.
-- There is no upper bound on the time we wait when waiting for a new SDU.
-- This makes it possible for miniprotocols to use timeouts that are larger
-- than 30s or wait forever.  `30s` for receiving an SDU corresponds to
-- a minimum speed limit of 17kbps.
--
-- ( 8      -- mux header length
-- + 0xffff -- maximum SDU payload
-- )
-- * 8
-- = 524_344 -- maximum bits in an SDU
--
--  524_344 / 30 / 1024 = 17kbps
--
sduTimeout :: DiffTime
sduTimeout :: DiffTime
sduTimeout = DiffTime
30

-- | For handshake, we put a limit of `10s` for sending or receiving a single
-- `MuxSDU`.
--
sduHandshakeTimeout :: DiffTime
sduHandshakeTimeout :: DiffTime
sduHandshakeTimeout = DiffTime
10

-- | Common arguments of various variants of `connectToNode`.
--
data ConnectToArgs fd addr vNumber vData = ConnectToArgs {
    forall fd addr vNumber vData.
ConnectToArgs fd addr vNumber vData
-> Codec (Handshake vNumber Term) DeserialiseFailure IO ByteString
ctaHandshakeCodec      :: Codec (Handshake vNumber CBOR.Term) CBOR.DeserialiseFailure IO BL.ByteString,
    forall fd addr vNumber vData.
ConnectToArgs fd addr vNumber vData
-> ProtocolTimeLimits (Handshake vNumber Term)
ctaHandshakeTimeLimits :: ProtocolTimeLimits (Handshake vNumber CBOR.Term),
    forall fd addr vNumber vData.
ConnectToArgs fd addr vNumber vData
-> VersionDataCodec Term vNumber vData
ctaVersionDataCodec    :: VersionDataCodec CBOR.Term vNumber vData,
    forall fd addr vNumber vData.
ConnectToArgs fd addr vNumber vData
-> NetworkConnectTracers addr vNumber
ctaConnectTracers      :: NetworkConnectTracers addr vNumber,
    forall fd addr vNumber vData.
ConnectToArgs fd addr vNumber vData -> HandshakeCallbacks vData
ctaHandshakeCallbacks  :: HandshakeCallbacks vData
  }


-- | Connect to a remote node.  It is using bracket to enclose the underlying
-- socket acquisition.  This implies that when the continuation exits the
-- underlying bearer will get closed.
--
-- The connection will start with handshake protocol sending @Versions@ to the
-- remote peer.  It must fit into @'maxTransmissionUnit'@ (~5k bytes).
--
-- Exceptions thrown by 'MuxApplication' are rethrown by 'connectToNode'.
connectToNode
  :: forall muxMode vNumber vData fd addr a b.
     ( Ord vNumber
     , Typeable vNumber
     , Show vNumber
     , Mx.HasInitiator muxMode ~ True
     )
  => Snocket IO fd addr
  -> Mx.MakeBearer IO fd
  -> ConnectToArgs fd addr vNumber vData
  -> (fd -> IO ()) -- ^ configure socket
  -> Versions vNumber vData (OuroborosApplicationWithMinimalCtx muxMode addr BL.ByteString IO a b)
  -> Maybe addr
  -- ^ local address; the created socket will bind to it
  -> addr
  -- ^ remote address
  -> IO (Either SomeException (Either a b))
connectToNode :: forall (muxMode :: Mode) vNumber vData fd addr a b.
(Ord vNumber, Typeable vNumber, Show vNumber,
 HasInitiator muxMode ~ 'True) =>
Snocket IO fd addr
-> MakeBearer IO fd
-> ConnectToArgs fd addr vNumber vData
-> (fd -> IO ())
-> Versions
     vNumber
     vData
     (OuroborosApplicationWithMinimalCtx muxMode addr ByteString IO a b)
-> Maybe addr
-> addr
-> IO (Either SomeException (Either a b))
connectToNode Snocket IO fd addr
sn MakeBearer IO fd
mkBearer ConnectToArgs fd addr vNumber vData
args fd -> IO ()
configureSock Versions
  vNumber
  vData
  (OuroborosApplicationWithMinimalCtx muxMode addr ByteString IO a b)
versions Maybe addr
localAddr addr
remoteAddr =
  Snocket IO fd addr
-> MakeBearer IO fd
-> ConnectToArgs fd addr vNumber vData
-> (fd -> IO ())
-> Versions
     vNumber
     vData
     (OuroborosApplicationWithMinimalCtx muxMode addr ByteString IO a b)
-> Maybe addr
-> addr
-> (ConnectionId addr
    -> vNumber
    -> vData
    -> OuroborosApplicationWithMinimalCtx
         muxMode addr ByteString IO a b
    -> Mux muxMode IO
    -> Async IO ()
    -> IO (Either SomeException (Either a b)))
-> IO (Either SomeException (Either a b))
forall (muxMode :: Mode) vNumber vData fd addr a b x.
(Ord vNumber, Typeable vNumber, Show vNumber,
 HasInitiator muxMode ~ 'True) =>
Snocket IO fd addr
-> MakeBearer IO fd
-> ConnectToArgs fd addr vNumber vData
-> (fd -> IO ())
-> Versions
     vNumber
     vData
     (OuroborosApplicationWithMinimalCtx muxMode addr ByteString IO a b)
-> Maybe addr
-> addr
-> (ConnectionId addr
    -> vNumber
    -> vData
    -> OuroborosApplicationWithMinimalCtx
         muxMode addr ByteString IO a b
    -> Mux muxMode IO
    -> Async IO ()
    -> IO x)
-> IO x
connectToNodeWithMux Snocket IO fd addr
sn MakeBearer IO fd
mkBearer ConnectToArgs fd addr vNumber vData
args fd -> IO ()
configureSock Versions
  vNumber
  vData
  (OuroborosApplicationWithMinimalCtx muxMode addr ByteString IO a b)
versions Maybe addr
localAddr addr
remoteAddr ConnectionId addr
-> vNumber
-> vData
-> OuroborosApplicationWithMinimalCtx
     muxMode addr ByteString IO a b
-> Mux muxMode IO
-> Async IO ()
-> IO (Either SomeException (Either a b))
forall (muxMode :: Mode) addr vNumber vData (m :: * -> *) a b.
(Alternative (STM m), MonadAsync m, MonadSTM m, MonadThrow m,
 MonadThrow (STM m)) =>
ConnectionId addr
-> vNumber
-> vData
-> OuroborosApplicationWithMinimalCtx muxMode addr ByteString m a b
-> Mux muxMode m
-> Async m ()
-> m (Either SomeException (Either a b))
simpleMuxCallback


-- | A version `connectToNode` which allows one to control which mini-protocols
-- to execute on a given connection.
connectToNodeWithMux
  :: forall muxMode vNumber vData fd addr a b x.
     ( Ord vNumber
     , Typeable vNumber
     , Show vNumber
     , Mx.HasInitiator muxMode ~ True
     )
  => Snocket IO fd addr
  -> Mx.MakeBearer IO fd
  -> ConnectToArgs fd addr vNumber vData
  -> (fd -> IO ()) -- ^ configure socket
  -> Versions vNumber vData (OuroborosApplicationWithMinimalCtx muxMode addr BL.ByteString IO a b)
  -- ^ application to run over the connection
  -- ^ remote address
  -> Maybe addr
  -> addr
  -> (    ConnectionId addr
       -> vNumber
       -> vData
       -> OuroborosApplicationWithMinimalCtx muxMode addr BL.ByteString IO a b
       -> Mx.Mux muxMode IO
       -> Async IO ()
       -> IO x)
  -- ^ callback which has access to ConnectionId, negotiated protocols, mux
  -- handle created for that connection and an `Async` handle to the thread
  -- which runs `Mx.runMux`.  The `Mux` handle allows schedule mini-protocols.
  --
  -- NOTE: when the callback returns or errors, the mux thread will be killed.
  -> IO x
connectToNodeWithMux :: forall (muxMode :: Mode) vNumber vData fd addr a b x.
(Ord vNumber, Typeable vNumber, Show vNumber,
 HasInitiator muxMode ~ 'True) =>
Snocket IO fd addr
-> MakeBearer IO fd
-> ConnectToArgs fd addr vNumber vData
-> (fd -> IO ())
-> Versions
     vNumber
     vData
     (OuroborosApplicationWithMinimalCtx muxMode addr ByteString IO a b)
-> Maybe addr
-> addr
-> (ConnectionId addr
    -> vNumber
    -> vData
    -> OuroborosApplicationWithMinimalCtx
         muxMode addr ByteString IO a b
    -> Mux muxMode IO
    -> Async IO ()
    -> IO x)
-> IO x
connectToNodeWithMux Snocket IO fd addr
sn MakeBearer IO fd
mkBearer ConnectToArgs fd addr vNumber vData
args fd -> IO ()
configureSock Versions
  vNumber
  vData
  (OuroborosApplicationWithMinimalCtx muxMode addr ByteString IO a b)
versions Maybe addr
localAddr addr
remoteAddr ConnectionId addr
-> vNumber
-> vData
-> OuroborosApplicationWithMinimalCtx
     muxMode addr ByteString IO a b
-> Mux muxMode IO
-> Async IO ()
-> IO x
k
  =
  IO fd -> (fd -> IO ()) -> (fd -> IO x) -> IO x
forall a b c. IO a -> (a -> IO b) -> (a -> IO c) -> IO c
forall (m :: * -> *) a b c.
MonadThrow m =>
m a -> (a -> m b) -> (a -> m c) -> m c
bracket
    (Snocket IO fd addr -> addr -> IO fd
forall (m :: * -> *) fd addr. Snocket m fd addr -> addr -> m fd
Snocket.openToConnect Snocket IO fd addr
sn addr
remoteAddr)
    (Snocket IO fd addr -> fd -> IO ()
forall (m :: * -> *) fd addr. Snocket m fd addr -> fd -> m ()
Snocket.close Snocket IO fd addr
sn)
    (\fd
sd -> do
      fd -> IO ()
configureSock fd
sd
      (addr -> IO ()) -> Maybe addr -> IO ()
forall (t :: * -> *) (f :: * -> *) a b.
(Foldable t, Applicative f) =>
(a -> f b) -> t a -> f ()
traverse_ (Snocket IO fd addr -> fd -> addr -> IO ()
forall (m :: * -> *) fd addr.
Snocket m fd addr -> fd -> addr -> m ()
Snocket.bind Snocket IO fd addr
sn fd
sd) Maybe addr
localAddr
      Snocket IO fd addr -> fd -> addr -> IO ()
forall (m :: * -> *) fd addr.
Snocket m fd addr -> fd -> addr -> m ()
Snocket.connect Snocket IO fd addr
sn fd
sd addr
remoteAddr
      Snocket IO fd addr
-> MakeBearer IO fd
-> ConnectToArgs fd addr vNumber vData
-> Versions
     vNumber
     vData
     (OuroborosApplicationWithMinimalCtx muxMode addr ByteString IO a b)
-> fd
-> (ConnectionId addr
    -> vNumber
    -> vData
    -> OuroborosApplicationWithMinimalCtx
         muxMode addr ByteString IO a b
    -> Mux muxMode IO
    -> Async IO ()
    -> IO x)
-> IO x
forall (muxMode :: Mode) vNumber vData fd addr a b x.
(Ord vNumber, Typeable vNumber, Show vNumber,
 HasInitiator muxMode ~ 'True) =>
Snocket IO fd addr
-> MakeBearer IO fd
-> ConnectToArgs fd addr vNumber vData
-> Versions
     vNumber
     vData
     (OuroborosApplicationWithMinimalCtx muxMode addr ByteString IO a b)
-> fd
-> (ConnectionId addr
    -> vNumber
    -> vData
    -> OuroborosApplicationWithMinimalCtx
         muxMode addr ByteString IO a b
    -> Mux muxMode IO
    -> Async IO ()
    -> IO x)
-> IO x
connectToNodeWithMux' Snocket IO fd addr
sn MakeBearer IO fd
mkBearer ConnectToArgs fd addr vNumber vData
args Versions
  vNumber
  vData
  (OuroborosApplicationWithMinimalCtx muxMode addr ByteString IO a b)
versions fd
sd ConnectionId addr
-> vNumber
-> vData
-> OuroborosApplicationWithMinimalCtx
     muxMode addr ByteString IO a b
-> Mux muxMode IO
-> Async IO ()
-> IO x
k
    )


-- | Connect to a remote node using an existing socket. It is up to to caller to
-- ensure that the socket is closed in case of an exception.
--
-- The connection will start with handshake protocol sending @Versions@ to the
-- remote peer.  It must fit into @'maxTransmissionUnit'@ (~5k bytes).
--
-- Exceptions thrown by @'MuxApplication'@ are rethrown by @'connectTo'@.
connectToNode'
  :: forall muxMode vNumber vData fd addr a b.
     ( Ord vNumber
     , Typeable vNumber
     , Show vNumber
     , Mx.HasInitiator muxMode ~ True
     )
  => Snocket IO fd addr
  -> Mx.MakeBearer IO fd
  -> ConnectToArgs fd addr vNumber vData
  -- ^ a configured socket to use to connect to a remote service provider
  -> Versions vNumber vData (OuroborosApplicationWithMinimalCtx muxMode addr BL.ByteString IO a b)
  -- ^ application to run over the connection
  -> fd
  -> IO (Either SomeException (Either a b))
connectToNode' :: forall (muxMode :: Mode) vNumber vData fd addr a b.
(Ord vNumber, Typeable vNumber, Show vNumber,
 HasInitiator muxMode ~ 'True) =>
Snocket IO fd addr
-> MakeBearer IO fd
-> ConnectToArgs fd addr vNumber vData
-> Versions
     vNumber
     vData
     (OuroborosApplicationWithMinimalCtx muxMode addr ByteString IO a b)
-> fd
-> IO (Either SomeException (Either a b))
connectToNode' Snocket IO fd addr
sn MakeBearer IO fd
mkBearer ConnectToArgs fd addr vNumber vData
args Versions
  vNumber
  vData
  (OuroborosApplicationWithMinimalCtx muxMode addr ByteString IO a b)
versions fd
as =
  Snocket IO fd addr
-> MakeBearer IO fd
-> ConnectToArgs fd addr vNumber vData
-> Versions
     vNumber
     vData
     (OuroborosApplicationWithMinimalCtx muxMode addr ByteString IO a b)
-> fd
-> (ConnectionId addr
    -> vNumber
    -> vData
    -> OuroborosApplicationWithMinimalCtx
         muxMode addr ByteString IO a b
    -> Mux muxMode IO
    -> Async IO ()
    -> IO (Either SomeException (Either a b)))
-> IO (Either SomeException (Either a b))
forall (muxMode :: Mode) vNumber vData fd addr a b x.
(Ord vNumber, Typeable vNumber, Show vNumber,
 HasInitiator muxMode ~ 'True) =>
Snocket IO fd addr
-> MakeBearer IO fd
-> ConnectToArgs fd addr vNumber vData
-> Versions
     vNumber
     vData
     (OuroborosApplicationWithMinimalCtx muxMode addr ByteString IO a b)
-> fd
-> (ConnectionId addr
    -> vNumber
    -> vData
    -> OuroborosApplicationWithMinimalCtx
         muxMode addr ByteString IO a b
    -> Mux muxMode IO
    -> Async IO ()
    -> IO x)
-> IO x
connectToNodeWithMux' Snocket IO fd addr
sn MakeBearer IO fd
mkBearer ConnectToArgs fd addr vNumber vData
args Versions
  vNumber
  vData
  (OuroborosApplicationWithMinimalCtx muxMode addr ByteString IO a b)
versions fd
as ConnectionId addr
-> vNumber
-> vData
-> OuroborosApplicationWithMinimalCtx
     muxMode addr ByteString IO a b
-> Mux muxMode IO
-> Async IO ()
-> IO (Either SomeException (Either a b))
forall (muxMode :: Mode) addr vNumber vData (m :: * -> *) a b.
(Alternative (STM m), MonadAsync m, MonadSTM m, MonadThrow m,
 MonadThrow (STM m)) =>
ConnectionId addr
-> vNumber
-> vData
-> OuroborosApplicationWithMinimalCtx muxMode addr ByteString m a b
-> Mux muxMode m
-> Async m ()
-> m (Either SomeException (Either a b))
simpleMuxCallback


connectToNodeWithMux'
  :: forall muxMode vNumber vData fd addr a b x.
     ( Ord vNumber
     , Typeable vNumber
     , Show vNumber
     , Mx.HasInitiator muxMode ~ True
     )
  => Snocket IO fd addr
  -> Mx.MakeBearer IO fd
  -> ConnectToArgs fd addr vNumber vData
  -> Versions vNumber vData (OuroborosApplicationWithMinimalCtx muxMode addr BL.ByteString IO a b)
  -- ^ application to run over the connection
  -- ^ a configured socket to use to connect to a remote service provider
  -> fd
  -> (    ConnectionId addr
       -> vNumber
       -> vData
       -> OuroborosApplicationWithMinimalCtx muxMode addr BL.ByteString IO a b
       -> Mx.Mux muxMode IO
       -> Async IO ()
       -> IO x)
  -- ^ callback which has access to ConnectionId, negotiated protocols, mux
  -- handle created for that connection and an `Async` handle to the thread
  -- which runs `Mx.runMux`.  The `Mux` handle allows schedule mini-protocols.
  --
  -- NOTE: when the callback returns or errors, the mux thread will be killed.
  -> IO x
connectToNodeWithMux' :: forall (muxMode :: Mode) vNumber vData fd addr a b x.
(Ord vNumber, Typeable vNumber, Show vNumber,
 HasInitiator muxMode ~ 'True) =>
Snocket IO fd addr
-> MakeBearer IO fd
-> ConnectToArgs fd addr vNumber vData
-> Versions
     vNumber
     vData
     (OuroborosApplicationWithMinimalCtx muxMode addr ByteString IO a b)
-> fd
-> (ConnectionId addr
    -> vNumber
    -> vData
    -> OuroborosApplicationWithMinimalCtx
         muxMode addr ByteString IO a b
    -> Mux muxMode IO
    -> Async IO ()
    -> IO x)
-> IO x
connectToNodeWithMux'
  Snocket IO fd addr
sn MakeBearer IO fd
makeBearer
  ConnectToArgs {
      ctaHandshakeCodec :: forall fd addr vNumber vData.
ConnectToArgs fd addr vNumber vData
-> Codec (Handshake vNumber Term) DeserialiseFailure IO ByteString
ctaHandshakeCodec      = Codec (Handshake vNumber Term) DeserialiseFailure IO ByteString
handshakeCodec,
      ctaHandshakeTimeLimits :: forall fd addr vNumber vData.
ConnectToArgs fd addr vNumber vData
-> ProtocolTimeLimits (Handshake vNumber Term)
ctaHandshakeTimeLimits = ProtocolTimeLimits (Handshake vNumber Term)
handshakeTimeLimits,
      ctaVersionDataCodec :: forall fd addr vNumber vData.
ConnectToArgs fd addr vNumber vData
-> VersionDataCodec Term vNumber vData
ctaVersionDataCodec    = VersionDataCodec Term vNumber vData
versionDataCodec,
      ctaConnectTracers :: forall fd addr vNumber vData.
ConnectToArgs fd addr vNumber vData
-> NetworkConnectTracers addr vNumber
ctaConnectTracers      =
        NetworkConnectTracers {
          Tracer IO (WithBearer (ConnectionId addr) Trace)
nctMuxTracer :: forall addr vNumber.
NetworkConnectTracers addr vNumber
-> Tracer IO (WithBearer (ConnectionId addr) Trace)
nctMuxTracer :: Tracer IO (WithBearer (ConnectionId addr) Trace)
nctMuxTracer,
          Tracer
  IO
  (WithBearer
     (ConnectionId addr) (TraceSendRecv (Handshake vNumber Term)))
nctHandshakeTracer :: forall addr vNumber.
NetworkConnectTracers addr vNumber
-> Tracer
     IO
     (WithBearer
        (ConnectionId addr) (TraceSendRecv (Handshake vNumber Term)))
nctHandshakeTracer :: Tracer
  IO
  (WithBearer
     (ConnectionId addr) (TraceSendRecv (Handshake vNumber Term)))
nctHandshakeTracer
        },
      ctaHandshakeCallbacks :: forall fd addr vNumber vData.
ConnectToArgs fd addr vNumber vData -> HandshakeCallbacks vData
ctaHandshakeCallbacks  = HandshakeCallbacks vData
handshakeCallbacks
  }
  Versions
  vNumber
  vData
  (OuroborosApplicationWithMinimalCtx muxMode addr ByteString IO a b)
versions fd
sd ConnectionId addr
-> vNumber
-> vData
-> OuroborosApplicationWithMinimalCtx
     muxMode addr ByteString IO a b
-> Mux muxMode IO
-> Async IO ()
-> IO x
k = do
    connectionId <- (\addr
localAddress addr
remoteAddress -> ConnectionId { addr
localAddress :: addr
localAddress :: addr
localAddress, addr
remoteAddress :: addr
remoteAddress :: addr
remoteAddress })
                (addr -> addr -> ConnectionId addr)
-> IO addr -> IO (addr -> ConnectionId addr)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Snocket IO fd addr -> fd -> IO addr
forall (m :: * -> *) fd addr. Snocket m fd addr -> fd -> m addr
Snocket.getLocalAddr Snocket IO fd addr
sn fd
sd IO (addr -> ConnectionId addr) -> IO addr -> IO (ConnectionId addr)
forall a b. IO (a -> b) -> IO a -> IO b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Snocket IO fd addr -> fd -> IO addr
forall (m :: * -> *) fd addr. Snocket m fd addr -> fd -> m addr
Snocket.getRemoteAddr Snocket IO fd addr
sn fd
sd
    muxTracer <- initDeltaQTracer' $ Mx.WithBearer connectionId `contramap` nctMuxTracer
    ts_start <- getMonotonicTime

    handshakeBearer <- Mx.getBearer makeBearer sduHandshakeTimeout muxTracer sd
    app_e <-
      runHandshakeClient
        handshakeBearer
        connectionId
        -- TODO: push 'HandshakeArguments' up the call stack.
        HandshakeArguments {
          haHandshakeTracer  = nctHandshakeTracer,
          haHandshakeCodec   = handshakeCodec,
          haVersionDataCodec = versionDataCodec,
          haAcceptVersion    = acceptCb handshakeCallbacks,
          haQueryVersion     = queryCb handshakeCallbacks,
          haTimeLimits       = handshakeTimeLimits
        }
        versions
    ts_end <- getMonotonicTime
    case app_e of
       Left (HandshakeProtocolLimit ProtocolLimitFailure
err) -> do
         Tracer IO Trace -> Trace -> IO ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer IO Trace
muxTracer (Trace -> IO ()) -> Trace -> IO ()
forall a b. (a -> b) -> a -> b
$ ProtocolLimitFailure -> DiffTime -> Trace
forall e. Exception e => e -> DiffTime -> Trace
Mx.TraceHandshakeClientError ProtocolLimitFailure
err (Time -> Time -> DiffTime
diffTime Time
ts_end Time
ts_start)
         ProtocolLimitFailure -> IO x
forall e a. Exception e => e -> IO a
forall (m :: * -> *) e a. (MonadThrow m, Exception e) => e -> m a
throwIO ProtocolLimitFailure
err

       Left (HandshakeProtocolError HandshakeProtocolError vNumber
err) -> do
         Tracer IO Trace -> Trace -> IO ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer IO Trace
muxTracer (Trace -> IO ()) -> Trace -> IO ()
forall a b. (a -> b) -> a -> b
$ HandshakeProtocolError vNumber -> DiffTime -> Trace
forall e. Exception e => e -> DiffTime -> Trace
Mx.TraceHandshakeClientError HandshakeProtocolError vNumber
err (Time -> Time -> DiffTime
diffTime Time
ts_end Time
ts_start)
         HandshakeProtocolError vNumber -> IO x
forall e a. Exception e => e -> IO a
forall (m :: * -> *) e a. (MonadThrow m, Exception e) => e -> m a
throwIO HandshakeProtocolError vNumber
err

       Right (HandshakeNegotiationResult OuroborosApplicationWithMinimalCtx muxMode addr ByteString IO a b
app vNumber
versionNumber vData
agreedOptions) -> do
         Tracer IO Trace -> Trace -> IO ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer IO Trace
muxTracer (Trace -> IO ()) -> Trace -> IO ()
forall a b. (a -> b) -> a -> b
$ DiffTime -> Trace
Mx.TraceHandshakeClientEnd (Time -> Time -> DiffTime
diffTime Time
ts_end Time
ts_start)
         bearer <- MakeBearer IO fd
-> DiffTime -> Tracer IO Trace -> fd -> IO (Bearer IO)
forall (m :: * -> *) fd.
MakeBearer m fd -> DiffTime -> Tracer m Trace -> fd -> m (Bearer m)
Mx.getBearer MakeBearer IO fd
makeBearer DiffTime
sduTimeout Tracer IO Trace
muxTracer fd
sd
         mux <- Mx.new (toMiniProtocolInfos (runForkPolicy noBindForkPolicy remoteAddress) app)
         withAsync (Mx.run muxTracer mux bearer) $ \Async IO ()
aid ->
           ConnectionId addr
-> vNumber
-> vData
-> OuroborosApplicationWithMinimalCtx
     muxMode addr ByteString IO a b
-> Mux muxMode IO
-> Async IO ()
-> IO x
k ConnectionId addr
connectionId vNumber
versionNumber vData
agreedOptions OuroborosApplicationWithMinimalCtx muxMode addr ByteString IO a b
app Mux muxMode IO
mux Async IO ()
aid

       Right (HandshakeQueryResult Map vNumber (Either Text vData)
_vMap) -> do
         Tracer IO Trace -> Trace -> IO ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer IO Trace
muxTracer (Trace -> IO ()) -> Trace -> IO ()
forall a b. (a -> b) -> a -> b
$ DiffTime -> Trace
Mx.TraceHandshakeClientEnd (Time -> Time -> DiffTime
diffTime Time
ts_end Time
ts_start)
         HandshakeProtocolError vNumber -> IO x
forall e a. Exception e => e -> IO a
forall (m :: * -> *) e a. (MonadThrow m, Exception e) => e -> m a
throwIO (forall vNumber. HandshakeProtocolError vNumber
QueryNotSupported @vNumber)


-- | An internal mux callback which starts all mini-protocols and blocks
-- until the first one terminates.  It returns the result (or error) of the
-- first terminated mini-protocol.
--
simpleMuxCallback
  :: forall muxMode addr vNumber vData m a b.
     ( Alternative (STM m)
     , MonadAsync m
     , MonadSTM   m
     , MonadThrow m
     , MonadThrow (STM m)
     )
  => ConnectionId addr
  -> vNumber
  -> vData
  -> OuroborosApplicationWithMinimalCtx muxMode addr BL.ByteString m a b
  -> Mx.Mux muxMode m
  -> Async m ()
  -> m (Either SomeException (Either a b))
simpleMuxCallback :: forall (muxMode :: Mode) addr vNumber vData (m :: * -> *) a b.
(Alternative (STM m), MonadAsync m, MonadSTM m, MonadThrow m,
 MonadThrow (STM m)) =>
ConnectionId addr
-> vNumber
-> vData
-> OuroborosApplicationWithMinimalCtx muxMode addr ByteString m a b
-> Mux muxMode m
-> Async m ()
-> m (Either SomeException (Either a b))
simpleMuxCallback ConnectionId addr
connectionId vNumber
_ vData
_ OuroborosApplicationWithMinimalCtx muxMode addr ByteString m a b
app Mux muxMode m
mux Async m ()
aid = do
    let initCtx :: MinimalInitiatorContext addr
initCtx = ConnectionId addr -> MinimalInitiatorContext addr
forall addr. ConnectionId addr -> MinimalInitiatorContext addr
MinimalInitiatorContext ConnectionId addr
connectionId
        respCtx :: ResponderContext addr
respCtx = ConnectionId addr -> ResponderContext addr
forall addr. ConnectionId addr -> ResponderContext addr
ResponderContext ConnectionId addr
connectionId

    resOps <- [m (STM m (Either SomeException (Either a b)))]
-> m [STM m (Either SomeException (Either a b))]
forall (t :: * -> *) (m :: * -> *) a.
(Traversable t, Monad m) =>
t (m a) -> m (t a)
forall (m :: * -> *) a. Monad m => [m a] -> m [a]
sequence
      [ Mux muxMode m
-> MiniProtocolNum
-> MiniProtocolDirection muxMode
-> StartOnDemandOrEagerly
-> (ByteChannel m -> m (Either a b, Maybe ByteString))
-> m (STM m (Either SomeException (Either a b)))
forall (mode :: Mode) (m :: * -> *) a.
(Alternative (STM m), MonadSTM m, MonadThrow m,
 MonadThrow (STM m)) =>
Mux mode m
-> MiniProtocolNum
-> MiniProtocolDirection mode
-> StartOnDemandOrEagerly
-> (ByteChannel m -> m (a, Maybe ByteString))
-> m (STM m (Either SomeException a))
Mx.runMiniProtocol
          Mux muxMode m
mux
          MiniProtocolNum
miniProtocolNum
          MiniProtocolDirection muxMode
miniProtocolDir
          StartOnDemandOrEagerly
Mx.StartEagerly
          ByteChannel m -> m (Either a b, Maybe ByteString)
action
      | MiniProtocol{MiniProtocolNum
miniProtocolNum :: MiniProtocolNum
miniProtocolNum :: forall (mode :: Mode) initiatorCtx responderCtx bytes (m :: * -> *)
       a b.
MiniProtocol mode initiatorCtx responderCtx bytes m a b
-> MiniProtocolNum
miniProtocolNum, RunMiniProtocol
  muxMode
  (MinimalInitiatorContext addr)
  (ResponderContext addr)
  ByteString
  m
  a
  b
miniProtocolRun :: RunMiniProtocol
  muxMode
  (MinimalInitiatorContext addr)
  (ResponderContext addr)
  ByteString
  m
  a
  b
miniProtocolRun :: forall (mode :: Mode) initiatorCtx responderCtx bytes (m :: * -> *)
       a b.
MiniProtocol mode initiatorCtx responderCtx bytes m a b
-> RunMiniProtocol mode initiatorCtx responderCtx bytes m a b
miniProtocolRun}
          <- OuroborosApplicationWithMinimalCtx muxMode addr ByteString m a b
-> [MiniProtocol
      muxMode
      (MinimalInitiatorContext addr)
      (ResponderContext addr)
      ByteString
      m
      a
      b]
forall (mode :: Mode) initiatorCtx responderCtx bytes (m :: * -> *)
       a b.
OuroborosApplication mode initiatorCtx responderCtx bytes m a b
-> [MiniProtocol mode initiatorCtx responderCtx bytes m a b]
getOuroborosApplication OuroborosApplicationWithMinimalCtx muxMode addr ByteString m a b
app
      , (MiniProtocolDirection muxMode
miniProtocolDir, ByteChannel m -> m (Either a b, Maybe ByteString)
action) <-
          case RunMiniProtocol
  muxMode
  (MinimalInitiatorContext addr)
  (ResponderContext addr)
  ByteString
  m
  a
  b
miniProtocolRun of
            InitiatorProtocolOnly MiniProtocolCb (MinimalInitiatorContext addr) ByteString m a
initiator ->
              [(MiniProtocolDirection 'InitiatorMode
Mx.InitiatorDirectionOnly, ((a, Maybe ByteString) -> (Either a b, Maybe ByteString))
-> m (a, Maybe ByteString) -> m (Either a b, Maybe ByteString)
forall a b. (a -> b) -> m a -> m b
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
fmap ((a -> Either a b)
-> (a, Maybe ByteString) -> (Either a b, Maybe ByteString)
forall a b c. (a -> b) -> (a, c) -> (b, c)
forall (p :: * -> * -> *) a b c.
Bifunctor p =>
(a -> b) -> p a c -> p b c
first a -> Either a b
forall a b. a -> Either a b
Left) (m (a, Maybe ByteString) -> m (Either a b, Maybe ByteString))
-> (ByteChannel m -> m (a, Maybe ByteString))
-> ByteChannel m
-> m (Either a b, Maybe ByteString)
forall b c a. (b -> c) -> (a -> b) -> a -> c
. MiniProtocolCb (MinimalInitiatorContext addr) ByteString m a
-> MinimalInitiatorContext addr
-> ByteChannel m
-> m (a, Maybe ByteString)
forall ctx bytes (m :: * -> *) a.
MiniProtocolCb ctx bytes m a
-> ctx -> Channel m bytes -> m (a, Maybe bytes)
runMiniProtocolCb MiniProtocolCb (MinimalInitiatorContext addr) ByteString m a
initiator MinimalInitiatorContext addr
initCtx)]
            ResponderProtocolOnly MiniProtocolCb (ResponderContext addr) ByteString m b
responder ->
              [(MiniProtocolDirection 'ResponderMode
Mx.ResponderDirectionOnly, ((b, Maybe ByteString) -> (Either a b, Maybe ByteString))
-> m (b, Maybe ByteString) -> m (Either a b, Maybe ByteString)
forall a b. (a -> b) -> m a -> m b
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
fmap ((b -> Either a b)
-> (b, Maybe ByteString) -> (Either a b, Maybe ByteString)
forall a b c. (a -> b) -> (a, c) -> (b, c)
forall (p :: * -> * -> *) a b c.
Bifunctor p =>
(a -> b) -> p a c -> p b c
first b -> Either a b
forall a b. b -> Either a b
Right) (m (b, Maybe ByteString) -> m (Either a b, Maybe ByteString))
-> (ByteChannel m -> m (b, Maybe ByteString))
-> ByteChannel m
-> m (Either a b, Maybe ByteString)
forall b c a. (b -> c) -> (a -> b) -> a -> c
. MiniProtocolCb (ResponderContext addr) ByteString m b
-> ResponderContext addr
-> ByteChannel m
-> m (b, Maybe ByteString)
forall ctx bytes (m :: * -> *) a.
MiniProtocolCb ctx bytes m a
-> ctx -> Channel m bytes -> m (a, Maybe bytes)
runMiniProtocolCb MiniProtocolCb (ResponderContext addr) ByteString m b
responder ResponderContext addr
respCtx)]
            InitiatorAndResponderProtocol MiniProtocolCb (MinimalInitiatorContext addr) ByteString m a
initiator MiniProtocolCb (ResponderContext addr) ByteString m b
responder ->
              [(MiniProtocolDirection 'InitiatorResponderMode
Mx.InitiatorDirection, ((a, Maybe ByteString) -> (Either a b, Maybe ByteString))
-> m (a, Maybe ByteString) -> m (Either a b, Maybe ByteString)
forall a b. (a -> b) -> m a -> m b
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
fmap ((a -> Either a b)
-> (a, Maybe ByteString) -> (Either a b, Maybe ByteString)
forall a b c. (a -> b) -> (a, c) -> (b, c)
forall (p :: * -> * -> *) a b c.
Bifunctor p =>
(a -> b) -> p a c -> p b c
first a -> Either a b
forall a b. a -> Either a b
Left) (m (a, Maybe ByteString) -> m (Either a b, Maybe ByteString))
-> (ByteChannel m -> m (a, Maybe ByteString))
-> ByteChannel m
-> m (Either a b, Maybe ByteString)
forall b c a. (b -> c) -> (a -> b) -> a -> c
. MiniProtocolCb (MinimalInitiatorContext addr) ByteString m a
-> MinimalInitiatorContext addr
-> ByteChannel m
-> m (a, Maybe ByteString)
forall ctx bytes (m :: * -> *) a.
MiniProtocolCb ctx bytes m a
-> ctx -> Channel m bytes -> m (a, Maybe bytes)
runMiniProtocolCb MiniProtocolCb (MinimalInitiatorContext addr) ByteString m a
initiator MinimalInitiatorContext addr
initCtx)
              ,(MiniProtocolDirection 'InitiatorResponderMode
Mx.ResponderDirection, ((b, Maybe ByteString) -> (Either a b, Maybe ByteString))
-> m (b, Maybe ByteString) -> m (Either a b, Maybe ByteString)
forall a b. (a -> b) -> m a -> m b
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
fmap ((b -> Either a b)
-> (b, Maybe ByteString) -> (Either a b, Maybe ByteString)
forall a b c. (a -> b) -> (a, c) -> (b, c)
forall (p :: * -> * -> *) a b c.
Bifunctor p =>
(a -> b) -> p a c -> p b c
first b -> Either a b
forall a b. b -> Either a b
Right) (m (b, Maybe ByteString) -> m (Either a b, Maybe ByteString))
-> (ByteChannel m -> m (b, Maybe ByteString))
-> ByteChannel m
-> m (Either a b, Maybe ByteString)
forall b c a. (b -> c) -> (a -> b) -> a -> c
. MiniProtocolCb (ResponderContext addr) ByteString m b
-> ResponderContext addr
-> ByteChannel m
-> m (b, Maybe ByteString)
forall ctx bytes (m :: * -> *) a.
MiniProtocolCb ctx bytes m a
-> ctx -> Channel m bytes -> m (a, Maybe bytes)
runMiniProtocolCb MiniProtocolCb (ResponderContext addr) ByteString m b
responder ResponderContext addr
respCtx)]
      ]

    -- Wait for the first MuxApplication to finish, then stop the mux.
    r <- waitOnAny resOps
    Mx.stop mux
    wait aid
    return r
  where
    waitOnAny :: [STM m (Either SomeException x)] -> m (Either SomeException x)
    waitOnAny :: forall x.
[STM m (Either SomeException x)] -> m (Either SomeException x)
waitOnAny = STM m (Either SomeException x) -> m (Either SomeException x)
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m (Either SomeException x) -> m (Either SomeException x))
-> ([STM m (Either SomeException x)]
    -> STM m (Either SomeException x))
-> [STM m (Either SomeException x)]
-> m (Either SomeException x)
forall b c a. (b -> c) -> (a -> b) -> a -> c
. FirstToFinish (STM m) (Either SomeException x)
-> STM m (Either SomeException x)
forall (m :: * -> *) a. FirstToFinish m a -> m a
runFirstToFinish (FirstToFinish (STM m) (Either SomeException x)
 -> STM m (Either SomeException x))
-> ([STM m (Either SomeException x)]
    -> FirstToFinish (STM m) (Either SomeException x))
-> [STM m (Either SomeException x)]
-> STM m (Either SomeException x)
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (STM m (Either SomeException x)
 -> FirstToFinish (STM m) (Either SomeException x))
-> [STM m (Either SomeException x)]
-> FirstToFinish (STM m) (Either SomeException x)
forall m a. Monoid m => (a -> m) -> [a] -> m
forall (t :: * -> *) m a.
(Foldable t, Monoid m) =>
(a -> m) -> t a -> m
foldMap STM m (Either SomeException x)
-> FirstToFinish (STM m) (Either SomeException x)
forall (m :: * -> *) a. m a -> FirstToFinish m a
FirstToFinish


-- Wraps a Socket inside a Snocket and calls connectToNode'
connectToNodeSocket
  :: forall muxMode vNumber vData a b.
     ( Ord vNumber
     , Typeable vNumber
     , Show vNumber
     , Mx.HasInitiator muxMode ~ True
     )
  => IOManager
  -> ConnectToArgs Socket.Socket Socket.SockAddr vNumber vData
  -> Versions vNumber vData (OuroborosApplicationWithMinimalCtx muxMode Socket.SockAddr BL.ByteString IO a b)
  -- ^ application to run over the connection
  -> Socket.Socket
  -> IO (Either SomeException (Either a b))
connectToNodeSocket :: forall (muxMode :: Mode) vNumber vData a b.
(Ord vNumber, Typeable vNumber, Show vNumber,
 HasInitiator muxMode ~ 'True) =>
IOManager
-> ConnectToArgs Socket SockAddr vNumber vData
-> Versions
     vNumber
     vData
     (OuroborosApplicationWithMinimalCtx
        muxMode SockAddr ByteString IO a b)
-> Socket
-> IO (Either SomeException (Either a b))
connectToNodeSocket IOManager
iocp ConnectToArgs Socket SockAddr vNumber vData
args Versions
  vNumber
  vData
  (OuroborosApplicationWithMinimalCtx
     muxMode SockAddr ByteString IO a b)
versions Socket
sd =
    Snocket IO Socket SockAddr
-> MakeBearer IO Socket
-> ConnectToArgs Socket SockAddr vNumber vData
-> Versions
     vNumber
     vData
     (OuroborosApplicationWithMinimalCtx
        muxMode SockAddr ByteString IO a b)
-> Socket
-> IO (Either SomeException (Either a b))
forall (muxMode :: Mode) vNumber vData fd addr a b.
(Ord vNumber, Typeable vNumber, Show vNumber,
 HasInitiator muxMode ~ 'True) =>
Snocket IO fd addr
-> MakeBearer IO fd
-> ConnectToArgs fd addr vNumber vData
-> Versions
     vNumber
     vData
     (OuroborosApplicationWithMinimalCtx muxMode addr ByteString IO a b)
-> fd
-> IO (Either SomeException (Either a b))
connectToNode'
      (IOManager -> Snocket IO Socket SockAddr
Snocket.socketSnocket IOManager
iocp)
      MakeBearer IO Socket
Mx.makeSocketBearer
      ConnectToArgs Socket SockAddr vNumber vData
args
      Versions
  vNumber
  vData
  (OuroborosApplicationWithMinimalCtx
     muxMode SockAddr ByteString IO a b)
versions
      Socket
sd

-- |
-- Wrapper for OuroborosResponderApplication and OuroborosInitiatorAndResponderApplication.
--
data SomeResponderApplication addr bytes m b where
     SomeResponderApplication
       :: forall muxMode addr bytes m a b.
          Mx.HasResponder muxMode ~ True
       => (OuroborosApplicationWithMinimalCtx muxMode addr bytes m a b)
       -> SomeResponderApplication addr bytes m b