{-# LANGUAGE LambdaCase        #-}
{-# LANGUAGE OverloadedStrings #-}

-- | High-level client implementation for NATS.
module Client.Implementation
  ( Server
  , ServerConfigError (..)
  , server
  , serverWithDefaultPort
  , serverHost
  , serverPort
  , connect
  , newClient
  , ConfigOption
  , withConnectName
  , withEcho
  , withAuthToken
  , withAuthTokenHandler
  , withUserPass
  , withUserPassHandler
  , withNKey
  , withNKeyHandler
  , withJWT
  , withJWTHandlers
  , withTLS
  , withTLSCert
  , withTLSRootCA
  , withTLSServerName
  , withTLSInsecure
  , withTLSConfigSource
  , withMinimumLogLevel
  , withLogAction
  , withConnectionAttempts
  , withConnectTimeoutMicros
  , withCallbackConcurrency
  , withMessageLimit
  , withPendingDeliveryLimits
  , withErrorHandler
  , withServerErrorHandler
  , withConnectionEventHandler
  , withBufferLimit
  , withExitAction
  , LogLevel (..)
  , LogEntry (..)
  , renderLogEntry
  , AuthTokenData
  , AuthTokenHandler
  , UserPassData
  , UserPassHandler
  , NKeyData
  , NKeyPublicKey
  , JWTTokenData
  , JWTHandler
  , SignatureHandler
  , AuthError (..)
  , TLSPublicKey
  , TLSPrivateKey
  , TLSCertData
  , TLSConfig (..)
  , TLSConfigSource
  , TLSConfigSourceFailure (..)
  , ClientExitReason (..)
  , ConnectionEvent (..)
  , ServerError
  , ServerErrorKind (..)
  , serverErrorReason
  , serverErrorKind
  , ConnectError (..)
  , ConnectAttemptError (..)
  , ConnectFailure (..)
  ) where

import           Auth.Config              (authMethods, mergeAuth)
import qualified Auth.Jwt                 as AuthJwt
import qualified Auth.NKey                as AuthNKey
import qualified Auth.None                as AuthNone
import qualified Auth.Token               as AuthToken
import           Auth.Types
    ( Auth
    , AuthError (..)
    , AuthTokenData
    , AuthTokenHandler
    , JWTHandler
    , JWTTokenData
    , NKeyData
    , NKeyPublicKey
    , SignatureHandler
    , UserPassData
    , UserPassHandler
    )
import qualified Auth.UserPass            as AuthUserPass
import           Client.API
    ( Client (..)
    , CloseConfig (..)
    , FlushConfig (..)
    , Message (..)
    , NatsError (..)
    , PingConfig (..)
    , RequestConfig (..)
    , ResetConfig (..)
    , Subscription (..)
    , UnsubscribeConfig (..)
    , subscriptionSid
    )
import           Control.Concurrent
    ( forkIOWithUnmask
    , killThread
    , myThreadId
    )
import           Control.Concurrent.STM
import           Control.Exception
    ( SomeException
    , finally
    , mask
    , mask_
    , onException
    )
import           Control.Monad            (forM_, void, when)
import qualified Data.ByteString          as BS
import qualified Data.ByteString.Char8    as BC
import           Data.Char                (isAsciiUpper)
import           Data.Maybe               (fromMaybe, isJust)
import           Data.Time.Clock          (NominalDiffTime)
import           Data.Version             (showVersion)
import           Engine
    ( closeClient
    , interruptConnection
    , resetClient
    , runEngine
    )
import           Lib.CallOption           (CallOption, applyCallOptions)
import           Lib.Logger
    ( LogEntry (..)
    , LogLevel (..)
    , LoggerConfig (..)
    , MonadLogger (..)
    , defaultLogger
    , newLogContext
    , renderLogEntry
    )
import           Network.Connection       (connectionApi)
import           Network.ConnectionAPI    (ConnectionAPI, newConn)
import qualified Network.ConnectionAPI    as Connection
import           Parser.Attoparsec        (parserApiWithMessageLimit)
import qualified Paths_natskell           as Package
import           Pipeline.Broadcasting    (broadcastingApi)
import           Pipeline.Streaming       (streamingApi)
import           Publish                  (defaultPublishConfig)
import           Publish.Config           (PublishConfig (..))
import           Queue.API
    ( QueueItem (QueueBatch, QueueConnectionScoped, QueueItem, QueuePayloadBound)
    , TryEnqueueResult (..)
    )
import           Queue.TransactionalQueue (newQueue)
import           State.Store
    ( ClientState
    , PingResult (..)
    , PublishEnqueueResult (..)
    , config
    , enqueue
    , enqueueOnGenerationTracked
    , enqueuePublishOnConnected
    , enqueuePublishOnConnectedGenerationTracked
    , isManagedThread
    , markClosed
    , newClientState
    , nextInbox
    , nextSid
    , readConnectionGeneration
    , readStatus
    , registerManagedThread
    , registerPingWaiterAndEnqueue
    , runClient
    , setClosing
    , setConnectName
    , startCallbackWorker
    , stopCallbackWorker
    , tryEnqueue
    , tryEnqueueOnGeneration
    , unregisterManagedThread
    , waitForCallbackWorker
    , waitForClosed
    , waitForConnected
    , waitForConnectionGenerationLoss
    , waitForInitialConnection
    , waitForNotRunning
    , withSubscriptionGate
    )
import           State.Types
    ( ClientConfig (..)
    , ClientExitReason (..)
    , ConnectAttemptError (..)
    , ConnectError (..)
    , ConnectFailure (..)
    , ConnectionEvent (..)
    , ConnectionState (..)
    , ServerError
    , ServerErrorKind (..)
    , TLSCertData
    , TLSConfig (..)
    , TLSConfigSource
    , TLSConfigSourceFailure (..)
    , TLSPrivateKey
    , TLSPublicKey
    , defaultTLSConfig
    , serverErrorKind
    , serverErrorReason
    )
import           Subscription.Store
    ( SubscriptionStore
    , awaitCallbackDrain
    , awaitNoTrackedExpiries
    , awaitSubscriptionDrain
    , closeStore
    , enqueueControl
    , hasTrackedExpiries
    , newSubscriptionStore
    , register
    , registerWithDispatchHooks
    , retire
    , startExpiryWorker
    , startWorkers
    , unregister
    )
import           Subscription.Types
    ( PendingLimits (..)
    , SubscribeConfig (..)
    , SubscriptionKind (..)
    , SubscriptionMeta (SubscriptionMeta)
    , defaultPendingLimits
    , isOneShotSubscription
    )
import           System.Timeout           (timeout)
import qualified Types.Connect            as Connect
import qualified Types.Msg                as Msg
import           Types.Ping               (Ping (..))
import qualified Types.Pub                as Pub
import qualified Types.Sub                as Sub
import qualified Types.Unsub              as Unsub
import           Validators.Validators    (validate)

data ClientOptions = ClientOptions
                       { ClientOptions -> Connect
optionConnectConfig :: Connect.Connect
                       , ClientOptions -> Auth
optionAuth :: Auth
                       , ClientOptions -> Maybe TLSConfig
optionTlsConfig :: Maybe TLSConfig
                       , ClientOptions -> Maybe TLSConfigSource
optionTlsConfigSource :: Maybe TLSConfigSource
                       , ClientOptions -> LoggerConfig
optionLoggerConfig :: LoggerConfig
                       , ClientOptions -> Int
optionConnectionAttempts :: Int
                       , ClientOptions -> Int
optionConnectTimeoutMicros :: Int
                       , ClientOptions -> Int
optionCallbackConcurrency :: Int
                       , ClientOptions -> Int
optionMessageLimit :: Int
                       , ClientOptions -> PendingLimits
optionPendingLimits :: PendingLimits
                       , ClientOptions -> NatsError -> IO ()
optionErrorHandler :: NatsError -> IO ()
                       , ClientOptions -> ServerError -> IO ()
optionServerErrorHandler :: ServerError -> IO ()
                       , ClientOptions -> ConnectionEvent -> IO ()
optionConnectionEventHandler :: ConnectionEvent -> IO ()
                       , ClientOptions -> ClientExitReason -> IO ()
optionExitAction :: ClientExitReason -> IO ()
                       , ClientOptions -> [([Char], Int)]
optionConnectOptions :: [(String, Int)]
                       }

data SubscriptionQueueResult = SubscriptionQueued | SubscriptionReconnect

data UnsubscribeQueueResult = UnsubscribeQueued | UnsubscribeReconnect | UnsubscribeReset

-- | An opaque NATS server endpoint.
--
-- Keeping this representation private allows endpoint schemes and transports
-- to be added without changing the connection API.
data Server = Server String Int
  deriving (Server -> Server -> Bool
(Server -> Server -> Bool)
-> (Server -> Server -> Bool) -> Eq Server
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: Server -> Server -> Bool
== :: Server -> Server -> Bool
$c/= :: Server -> Server -> Bool
/= :: Server -> Server -> Bool
Eq, Int -> Server -> ShowS
[Server] -> ShowS
Server -> [Char]
(Int -> Server -> ShowS)
-> (Server -> [Char]) -> ([Server] -> ShowS) -> Show Server
forall a.
(Int -> a -> ShowS) -> (a -> [Char]) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> Server -> ShowS
showsPrec :: Int -> Server -> ShowS
$cshow :: Server -> [Char]
show :: Server -> [Char]
$cshowList :: [Server] -> ShowS
showList :: [Server] -> ShowS
Show)

data ServerConfigError = EmptyServerHost
                       | InvalidServerPort Int
  deriving (ServerConfigError -> ServerConfigError -> Bool
(ServerConfigError -> ServerConfigError -> Bool)
-> (ServerConfigError -> ServerConfigError -> Bool)
-> Eq ServerConfigError
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: ServerConfigError -> ServerConfigError -> Bool
== :: ServerConfigError -> ServerConfigError -> Bool
$c/= :: ServerConfigError -> ServerConfigError -> Bool
/= :: ServerConfigError -> ServerConfigError -> Bool
Eq, Int -> ServerConfigError -> ShowS
[ServerConfigError] -> ShowS
ServerConfigError -> [Char]
(Int -> ServerConfigError -> ShowS)
-> (ServerConfigError -> [Char])
-> ([ServerConfigError] -> ShowS)
-> Show ServerConfigError
forall a.
(Int -> a -> ShowS) -> (a -> [Char]) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> ServerConfigError -> ShowS
showsPrec :: Int -> ServerConfigError -> ShowS
$cshow :: ServerConfigError -> [Char]
show :: ServerConfigError -> [Char]
$cshowList :: [ServerConfigError] -> ShowS
showList :: [ServerConfigError] -> ShowS
Show)

-- | Construct a TCP NATS server endpoint.
server :: String -> Int -> Either ServerConfigError Server
server :: [Char] -> Int -> Either ServerConfigError Server
server [Char]
host Int
port
  | [Char] -> Bool
forall a. [a] -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null [Char]
host = ServerConfigError -> Either ServerConfigError Server
forall a b. a -> Either a b
Left ServerConfigError
EmptyServerHost
  | Int
port Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
< Int
1 Bool -> Bool -> Bool
|| Int
port Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
> Int
65535 = ServerConfigError -> Either ServerConfigError Server
forall a b. a -> Either a b
Left (Int -> ServerConfigError
InvalidServerPort Int
port)
  | Bool
otherwise = Server -> Either ServerConfigError Server
forall a b. b -> Either a b
Right ([Char] -> Int -> Server
Server [Char]
host Int
port)

-- | Construct a server using the standard NATS port, 4222.
serverWithDefaultPort :: String -> Either ServerConfigError Server
serverWithDefaultPort :: [Char] -> Either ServerConfigError Server
serverWithDefaultPort [Char]
host = [Char] -> Int -> Either ServerConfigError Server
server [Char]
host Int
4222

serverHost :: Server -> String
serverHost :: Server -> [Char]
serverHost (Server [Char]
host Int
_) = [Char]
host

serverPort :: Server -> Int
serverPort :: Server -> Int
serverPort (Server [Char]
_ Int
port) = Int
port

-- | Connect to one of the configured NATS servers.
connect :: [Server] -> [ConfigOption] -> IO (Either ConnectError Client)
connect :: [Server] -> [ConfigOption] -> IO (Either ConnectError Client)
connect [Server]
servers =
  [([Char], Int)]
-> [ConfigOption] -> IO (Either ConnectError Client)
newClient [(Server -> [Char]
serverHost Server
endpoint, Server -> Int
serverPort Server
endpoint) | Server
endpoint <- [Server]
servers]

-- | Compatibility connection entry point using raw @(host, port)@ tuples.
newClient :: [(String, Int)] -> [ConfigOption] -> IO (Either ConnectError Client)
newClient :: [([Char], Int)]
-> [ConfigOption] -> IO (Either ConnectError Client)
newClient [([Char], Int)]
servers [ConfigOption]
configOptions = ((forall a. IO a -> IO a) -> IO (Either ConnectError Client))
-> IO (Either ConnectError Client)
forall b. ((forall a. IO a -> IO a) -> IO b) -> IO b
mask (((forall a. IO a -> IO a) -> IO (Either ConnectError Client))
 -> IO (Either ConnectError Client))
-> ((forall a. IO a -> IO a) -> IO (Either ConnectError Client))
-> IO (Either ConnectError Client)
forall a b. (a -> b) -> a -> b
$ \forall a. IO a -> IO a
restoreInitialWait -> do
  LoggerConfig
loggerConfig' <- IO LoggerConfig
defaultLogger
  TVar LogContext
ctx <- IO (TVar LogContext)
newLogContext
  let defaultOptions :: ClientOptions
defaultOptions = [ConfigOption] -> ConfigOption
forall a. [CallOption a] -> CallOption a
applyCallOptions [ConfigOption]
configOptions ClientOptions
        { optionConnectConfig :: Connect
optionConnectConfig = Connect
defaultConnect
        , optionAuth :: Auth
optionAuth = Auth
AuthNone.auth
        , optionTlsConfig :: Maybe TLSConfig
optionTlsConfig = Maybe TLSConfig
forall a. Maybe a
Nothing
        , optionTlsConfigSource :: Maybe TLSConfigSource
optionTlsConfigSource = Maybe TLSConfigSource
forall a. Maybe a
Nothing
        , optionLoggerConfig :: LoggerConfig
optionLoggerConfig = LoggerConfig
loggerConfig'
        , optionConnectionAttempts :: Int
optionConnectionAttempts = Int
5
        , optionConnectTimeoutMicros :: Int
optionConnectTimeoutMicros = Int
2 Int -> Int -> Int
forall a. Num a => a -> a -> a
* Int
1000000
        , optionCallbackConcurrency :: Int
optionCallbackConcurrency = Int
1
        , optionMessageLimit :: Int
optionMessageLimit = Int
1024 Int -> Int -> Int
forall a. Num a => a -> a -> a
* Int
1024
        , optionPendingLimits :: PendingLimits
optionPendingLimits = PendingLimits
defaultPendingLimits
        , optionErrorHandler :: NatsError -> IO ()
optionErrorHandler = IO () -> NatsError -> IO ()
forall a b. a -> b -> a
const (() -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ())
        , optionServerErrorHandler :: ServerError -> IO ()
optionServerErrorHandler = IO () -> ServerError -> IO ()
forall a b. a -> b -> a
const (() -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ())
        , optionConnectionEventHandler :: ConnectionEvent -> IO ()
optionConnectionEventHandler = IO () -> ConnectionEvent -> IO ()
forall a b. a -> b -> a
const (() -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ())
        , optionExitAction :: ClientExitReason -> IO ()
optionExitAction = IO () -> ClientExitReason -> IO ()
forall a b. a -> b -> a
const (() -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ())
        , optionConnectOptions :: [([Char], Int)]
optionConnectOptions = [([Char], Int)]
servers
        }
      clientConfig :: ClientConfig
clientConfig =
        ClientConfig
          { connectionAttempts :: Int
connectionAttempts = ClientOptions -> Int
optionConnectionAttempts ClientOptions
defaultOptions
          , connectTimeoutMicros :: Int
connectTimeoutMicros = ClientOptions -> Int
optionConnectTimeoutMicros ClientOptions
defaultOptions
          , callbackConcurrency :: Int
callbackConcurrency = ClientOptions -> Int
optionCallbackConcurrency ClientOptions
defaultOptions
          , messageLimit :: Int
messageLimit = ClientOptions -> Int
optionMessageLimit ClientOptions
defaultOptions
          , connectConfig :: Connect
connectConfig = ClientOptions -> Connect
optionConnectConfig ClientOptions
defaultOptions
          , loggerConfig :: LoggerConfig
loggerConfig = ClientOptions -> LoggerConfig
optionLoggerConfig ClientOptions
defaultOptions
          , tlsConfig :: Maybe TLSConfig
tlsConfig = ClientOptions -> Maybe TLSConfig
optionTlsConfig ClientOptions
defaultOptions
          , tlsConfigSource :: Maybe TLSConfigSource
tlsConfigSource = ClientOptions -> Maybe TLSConfigSource
optionTlsConfigSource ClientOptions
defaultOptions
          , serverErrorHandler :: ServerError -> IO ()
serverErrorHandler = ClientOptions -> ServerError -> IO ()
optionServerErrorHandler ClientOptions
defaultOptions
          , connectionEventHandler :: ConnectionEvent -> IO ()
connectionEventHandler = ClientOptions -> ConnectionEvent -> IO ()
optionConnectionEventHandler ClientOptions
defaultOptions
          , exitAction :: ClientExitReason -> IO ()
exitAction = ClientOptions -> ClientExitReason -> IO ()
optionExitAction ClientOptions
defaultOptions
          , connectOptions :: [([Char], Int)]
connectOptions = ClientOptions -> [([Char], Int)]
optionConnectOptions ClientOptions
defaultOptions
          }
      configuredAuth :: Auth
configuredAuth = ClientOptions -> Auth
optionAuth ClientOptions
defaultOptions

  Queue
queue <- IO Queue
newQueue
  Conn
conn <- ConnectionAPI -> IO Conn
newConn ConnectionAPI
connectionApi
  ClientState
clientState <- ClientConfig -> Queue -> Conn -> TVar LogContext -> IO ClientState
newClientState ClientConfig
clientConfig Queue
queue Conn
conn TVar LogContext
ctx
  SubscriptionStore
store <-
    PendingLimits -> IO () -> IO SubscriptionStore
newSubscriptionStore
      (ClientOptions -> PendingLimits
optionPendingLimits ClientOptions
defaultOptions)
      (ClientState -> (NatsError -> IO ()) -> IO ()
handleSlowConsumer ClientState
clientState (ClientOptions -> NatsError -> IO ()
optionErrorHandler ClientOptions
defaultOptions))

  ClientState -> Maybe SID -> IO ()
setConnectName ClientState
clientState (Connect -> Maybe SID
Connect.name (ClientOptions -> Connect
optionConnectConfig ClientOptions
defaultOptions))
  ClientState -> ClientOptions -> IO ()
logStaticConfiguration ClientState
clientState ClientOptions
defaultOptions

  [ThreadId]
callbackThreads <- Int
-> SubscriptionStore
-> STM ()
-> (SomeException -> IO ())
-> IO [ThreadId]
startWorkers
    (ClientConfig -> Int
callbackConcurrency ClientConfig
clientConfig)
    SubscriptionStore
store
    (do
        ClientState -> STM ()
waitForClosed ClientState
clientState
        SubscriptionStore -> STM ()
awaitCallbackDrain SubscriptionStore
store
        SubscriptionStore -> STM ()
awaitNoTrackedExpiries SubscriptionStore
store)
    (ClientState -> SomeException -> IO ()
handleCallbackError ClientState
clientState)

  ThreadId
expiryThread <- SubscriptionStore -> IO Bool -> IO ThreadId
startExpiryWorker SubscriptionStore
store (IO Bool -> IO ThreadId) -> IO Bool -> IO ThreadId
forall a b. (a -> b) -> a -> b
$
    ClientState -> SubscriptionStore -> IO Bool
shouldStopExpiryWorker ClientState
clientState SubscriptionStore
store

  (ThreadId -> IO ()) -> [ThreadId] -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
(a -> m b) -> t a -> m ()
mapM_ (ClientState -> ThreadId -> IO ()
registerManagedThread ClientState
clientState) (ThreadId
expiryThread ThreadId -> [ThreadId] -> [ThreadId]
forall a. a -> [a] -> [a]
: [ThreadId]
callbackThreads)
  CallbackWorker
callbackWorker <- ClientState -> IO CallbackWorker
startCallbackWorker ClientState
clientState

  TMVar ()
engineDone <- IO (TMVar ())
forall a. IO (TMVar a)
newEmptyTMVarIO
  ThreadId
engineThread <- ((forall a. IO a -> IO a) -> IO ()) -> IO ThreadId
forkIOWithUnmask (((forall a. IO a -> IO a) -> IO ()) -> IO ThreadId)
-> ((forall a. IO a -> IO a) -> IO ()) -> IO ThreadId
forall a b. (a -> b) -> a -> b
$ \forall a. IO a -> IO a
unmask ->
    do
      ThreadId
threadId <- IO ThreadId
myThreadId
      ClientState -> ThreadId -> IO ()
registerManagedThread ClientState
clientState ThreadId
threadId
      IO () -> IO ()
forall a. IO a -> IO a
unmask
        ( ConnectionAPI
-> StreamingAPI
-> BroadcastingAPI
-> ParserAPI ParsedMessage
-> ClientState
-> SubscriptionStore
-> Auth
-> IO ()
runEngine
            ConnectionAPI
connectionApi
            StreamingAPI
streamingApi
            BroadcastingAPI
broadcastingApi
            (Int -> ParserAPI ParsedMessage
parserApiWithMessageLimit (ClientConfig -> Int
messageLimit ClientConfig
clientConfig))
            ClientState
clientState
            SubscriptionStore
store
            Auth
configuredAuth
        )
        IO () -> IO () -> IO ()
forall a b. IO a -> IO b -> IO a
`finally` do
          ClientState -> ThreadId -> IO ()
unregisterManagedThread ClientState
clientState ThreadId
threadId
          STM () -> IO ()
forall a. STM a -> IO a
atomically (STM Bool -> STM ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (TMVar () -> () -> STM Bool
forall a. TMVar a -> a -> STM Bool
tryPutTMVar TMVar ()
engineDone ()))

  let auxiliaryThreads :: [ThreadId]
auxiliaryThreads = ThreadId
expiryThread ThreadId -> [ThreadId] -> [ThreadId]
forall a. a -> [a] -> [a]
: [ThreadId]
callbackThreads
      stopAuxiliaryThreads :: ThreadId -> IO ()
stopAuxiliaryThreads ThreadId
excluded =
        (ThreadId -> IO ()) -> [ThreadId] -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
(a -> m b) -> t a -> m ()
mapM_ ThreadId -> IO ()
killThread ((ThreadId -> Bool) -> [ThreadId] -> [ThreadId]
forall a. (a -> Bool) -> [a] -> [a]
filter (ThreadId -> ThreadId -> Bool
forall a. Eq a => a -> a -> Bool
/= ThreadId
excluded) [ThreadId]
auxiliaryThreads)
      finishClose :: ThreadId -> IO ()
finishClose ThreadId
excluded = do
        STM () -> IO ()
forall a. STM a -> IO a
atomically (TMVar () -> STM ()
forall a. TMVar a -> STM a
readTMVar TMVar ()
engineDone)
        ConnectionAPI -> Conn -> IO ()
Connection.close ConnectionAPI
connectionApi Conn
conn
        STM () -> IO ()
forall a. STM a -> IO a
atomically (SubscriptionStore -> STM ()
awaitCallbackDrain SubscriptionStore
store)
        ThreadId -> IO ()
stopAuxiliaryThreads ThreadId
excluded
        STM () -> IO ()
forall a. STM a -> IO a
atomically (CallbackWorker -> STM ()
waitForCallbackWorker CallbackWorker
callbackWorker)
      scheduleFinish :: ThreadId -> IO ()
scheduleFinish ThreadId
excluded = do
        ThreadId
_ <- ((forall a. IO a -> IO a) -> IO ()) -> IO ThreadId
forkIOWithUnmask (((forall a. IO a -> IO a) -> IO ()) -> IO ThreadId)
-> ((forall a. IO a -> IO a) -> IO ()) -> IO ThreadId
forall a b. (a -> b) -> a -> b
$ \forall a. IO a -> IO a
unmask -> IO () -> IO ()
forall a. IO a -> IO a
unmask (ThreadId -> IO ()
finishClose ThreadId
excluded)
        () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
      closeAndFinish :: IO ()
closeAndFinish = ((forall a. IO a -> IO a) -> IO ()) -> IO ()
forall b. ((forall a. IO a -> IO a) -> IO b) -> IO b
mask (((forall a. IO a -> IO a) -> IO ()) -> IO ())
-> ((forall a. IO a -> IO a) -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \forall a. IO a -> IO a
restore -> do
        ThreadId
current <- IO ThreadId
myThreadId
        ConnectionAPI -> ClientState -> SubscriptionStore -> IO ()
closeClient ConnectionAPI
connectionApi ClientState
clientState SubscriptionStore
store
          IO () -> IO () -> IO ()
forall a b. IO a -> IO b -> IO a
`onException` ThreadId -> IO ()
scheduleFinish ThreadId
current
        Bool
managed <- ClientState -> ThreadId -> IO Bool
isManagedThread ClientState
clientState ThreadId
current
        if Bool
managed
          then ThreadId -> IO ()
scheduleFinish ThreadId
current
          else IO () -> IO ()
forall a. IO a -> IO a
restore (ThreadId -> IO ()
finishClose ThreadId
current) IO () -> IO () -> IO ()
forall a b. IO a -> IO b -> IO a
`onException` ThreadId -> IO ()
scheduleFinish ThreadId
current

  let client :: Client
client = Client
        { publish :: SID -> SID -> [PublishOption] -> IO (Either NatsError ())
publish = \SID
subject SID
payload [PublishOption]
publishOptions -> do
            let cfg :: PublishConfig
cfg = [PublishOption] -> PublishOption
forall a. [CallOption a] -> CallOption a
applyCallOptions [PublishOption]
publishOptions PublishConfig
defaultPublishConfig
            ClientState
-> (NatsError -> IO ())
-> SID
-> SID
-> PublishConfig
-> IO (Either NatsError ())
publishClient
              ClientState
clientState
              (SubscriptionStore -> IO () -> IO ()
enqueueControl SubscriptionStore
store (IO () -> IO ()) -> (NatsError -> IO ()) -> NatsError -> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. ClientOptions -> NatsError -> IO ()
optionErrorHandler ClientOptions
defaultOptions)
              SID
subject
              SID
payload
              PublishConfig
cfg
        , subscribe :: SID
-> [SubscribeOption]
-> (Message -> IO ())
-> IO (Either NatsError Subscription)
subscribe = \SID
subject [SubscribeOption]
subscribeOptions Message -> IO ()
callback -> do
            let cfg :: SubscribeConfig
cfg = [SubscribeOption] -> SubscribeOption
forall a. [CallOption a] -> CallOption a
applyCallOptions [SubscribeOption]
subscribeOptions SubscribeConfig
defaultSubscribeConfig
            ClientState
-> SubscriptionStore
-> SubscriptionKind
-> SID
-> SubscribeConfig
-> (Message -> IO ())
-> IO (Either NatsError Subscription)
subscribeClient ClientState
clientState SubscriptionStore
store SubscriptionKind
StandardSubscription SID
subject SubscribeConfig
cfg Message -> IO ()
callback
        , subscribeOnce :: SID
-> [SubscribeOption]
-> (Message -> IO ())
-> IO (Either NatsError Subscription)
subscribeOnce = \SID
subject [SubscribeOption]
subscribeOptions Message -> IO ()
callback -> do
            let cfg :: SubscribeConfig
cfg = [SubscribeOption] -> SubscribeOption
forall a. [CallOption a] -> CallOption a
applyCallOptions [SubscribeOption]
subscribeOptions SubscribeConfig
defaultSubscribeConfig
            ClientState
-> SubscriptionStore
-> SubscriptionKind
-> SID
-> SubscribeConfig
-> (Message -> IO ())
-> IO (Either NatsError Subscription)
subscribeClient ClientState
clientState SubscriptionStore
store SubscriptionKind
OneShotSubscription SID
subject SubscribeConfig
cfg Message -> IO ()
callback
        , request :: SID -> SID -> [RequestOption] -> IO (Either NatsError Message)
request = \SID
subject SID
payload [RequestOption]
requestOptions -> do
            let cfg :: RequestConfig
cfg = [RequestOption] -> RequestOption
forall a. [CallOption a] -> CallOption a
applyCallOptions [RequestOption]
requestOptions RequestConfig
defaultRequestConfig
            ClientState
-> SubscriptionStore
-> SID
-> SID
-> RequestConfig
-> IO (Either NatsError Message)
requestClient ClientState
clientState SubscriptionStore
store SID
subject SID
payload RequestConfig
cfg
        , unsubscribe :: Subscription -> [UnsubscribeOption] -> IO (Either NatsError ())
unsubscribe = \Subscription
subscription [UnsubscribeOption]
options ->
            case [UnsubscribeOption] -> UnsubscribeOption
forall a. [CallOption a] -> CallOption a
applyCallOptions [UnsubscribeOption]
options UnsubscribeConfig
UnsubscribeConfig of
              UnsubscribeConfig
UnsubscribeConfig -> ClientState
-> SubscriptionStore -> Subscription -> IO (Either NatsError ())
unsubscribeClient ClientState
clientState SubscriptionStore
store Subscription
subscription
        , drainSubscriptions :: [Subscription] -> [FlushOption] -> IO (Either NatsError ())
drainSubscriptions = \[Subscription]
subscriptions [FlushOption]
options ->
            case [FlushOption] -> FlushOption
forall a. [CallOption a] -> CallOption a
applyCallOptions [FlushOption]
options FlushConfig
defaultFlushConfig of
              FlushConfig
FlushConfig ->
                ConnectionAPI
-> ClientState
-> SubscriptionStore
-> [Subscription]
-> NominalDiffTime
-> IO (Either NatsError ())
drainSubscriptionsClient
                  ConnectionAPI
connectionApi
                  ClientState
clientState
                  SubscriptionStore
store
                  [Subscription]
subscriptions
                  NominalDiffTime
defaultRoundTripTimeout
              FlushConfigTimeout NominalDiffTime
timeoutSeconds ->
                ConnectionAPI
-> ClientState
-> SubscriptionStore
-> [Subscription]
-> NominalDiffTime
-> IO (Either NatsError ())
drainSubscriptionsClient
                  ConnectionAPI
connectionApi
                  ClientState
clientState
                  SubscriptionStore
store
                  [Subscription]
subscriptions
                  NominalDiffTime
timeoutSeconds
        , newInbox :: IO SID
newInbox = ClientState -> IO SID
nextInbox ClientState
clientState
        , ping :: [PingOption] -> IO (Either NatsError ())
ping = \[PingOption]
options ->
            case [PingOption] -> PingOption
forall a. [CallOption a] -> CallOption a
applyCallOptions [PingOption]
options PingConfig
defaultPingConfig of
              PingConfig
PingConfig -> ConnectionAPI
-> ClientState -> NominalDiffTime -> IO (Either NatsError ())
flushClient ConnectionAPI
connectionApi ClientState
clientState NominalDiffTime
defaultRoundTripTimeout
              PingConfigTimeout NominalDiffTime
timeoutSeconds ->
                ConnectionAPI
-> ClientState -> NominalDiffTime -> IO (Either NatsError ())
flushClient ConnectionAPI
connectionApi ClientState
clientState NominalDiffTime
timeoutSeconds
        , flush :: [FlushOption] -> IO (Either NatsError ())
flush = \[FlushOption]
options ->
            case [FlushOption] -> FlushOption
forall a. [CallOption a] -> CallOption a
applyCallOptions [FlushOption]
options FlushConfig
defaultFlushConfig of
              FlushConfig
FlushConfig -> ConnectionAPI
-> ClientState -> NominalDiffTime -> IO (Either NatsError ())
flushClient ConnectionAPI
connectionApi ClientState
clientState NominalDiffTime
defaultRoundTripTimeout
              FlushConfigTimeout NominalDiffTime
timeoutSeconds ->
                ConnectionAPI
-> ClientState -> NominalDiffTime -> IO (Either NatsError ())
flushClient ConnectionAPI
connectionApi ClientState
clientState NominalDiffTime
timeoutSeconds
        , connectionState :: IO ConnectionState
connectionState = ClientState -> IO ConnectionState
readStatus ClientState
clientState
        , reset :: [ResetOption] -> IO ()
reset = \[ResetOption]
options ->
            case [ResetOption] -> ResetOption
forall a. [CallOption a] -> CallOption a
applyCallOptions [ResetOption]
options ResetConfig
ResetConfig of
              ResetConfig
ResetConfig -> ConnectionAPI -> ClientState -> SubscriptionStore -> IO ()
resetClient ConnectionAPI
connectionApi ClientState
clientState SubscriptionStore
store
        , close :: [CloseOption] -> IO ()
close = \[CloseOption]
options ->
            case [CloseOption] -> CloseOption
forall a. [CallOption a] -> CallOption a
applyCallOptions [CloseOption]
options CloseConfig
CloseConfig of
              CloseConfig
CloseConfig -> IO ()
closeAndFinish
        }

  let abortInitialWait :: IO ()
abortInitialWait = do
        ClientState -> ClientExitReason -> IO ()
setClosing ClientState
clientState ClientExitReason
ExitClosedByUser
        SubscriptionStore -> IO ()
closeStore SubscriptionStore
store
        IO Int -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (ConnectionAPI -> ClientState -> IO Int
interruptConnection ConnectionAPI
connectionApi ClientState
clientState)
        ThreadId -> IO ()
killThread ThreadId
engineThread
        STM () -> IO ()
forall a. STM a -> IO a
atomically (TMVar () -> STM ()
forall a. TMVar a -> STM a
readTMVar TMVar ()
engineDone)
        ConnectionAPI -> Conn -> IO ()
Connection.close ConnectionAPI
connectionApi Conn
conn
        IO (Maybe ClientExitReason) -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (ClientState -> ClientExitReason -> IO (Maybe ClientExitReason)
markClosed ClientState
clientState ClientExitReason
ExitClosedByUser)
        CallbackWorker -> IO ()
stopCallbackWorker CallbackWorker
callbackWorker
        ThreadId
current <- IO ThreadId
myThreadId
        ThreadId -> IO ()
stopAuxiliaryThreads ThreadId
current
  Either ConnectError ()
initialResult <-
    IO (Either ConnectError ()) -> IO (Either ConnectError ())
forall a. IO a -> IO a
restoreInitialWait
      (STM (Either ConnectError ()) -> IO (Either ConnectError ())
forall a. STM a -> IO a
atomically (STM (Either ConnectError ()) -> IO (Either ConnectError ()))
-> STM (Either ConnectError ()) -> IO (Either ConnectError ())
forall a b. (a -> b) -> a -> b
$
        ClientState -> STM (Either ConnectError ())
waitForInitialConnection ClientState
clientState
          STM (Either ConnectError ())
-> STM (Either ConnectError ()) -> STM (Either ConnectError ())
forall a. STM a -> STM a -> STM a
`orElse` (ConnectError -> Either ConnectError ()
forall a b. a -> Either a b
Left ([ConnectAttemptError] -> ConnectError
ConnectAttemptsExhausted []) Either ConnectError () -> STM () -> STM (Either ConnectError ())
forall a b. a -> STM b -> STM a
forall (f :: * -> *) a b. Functor f => a -> f b -> f a
<$ TMVar () -> STM ()
forall a. TMVar a -> STM a
readTMVar TMVar ()
engineDone))
      IO (Either ConnectError ()) -> IO () -> IO (Either ConnectError ())
forall a b. IO a -> IO b -> IO a
`onException` IO ()
abortInitialWait
  case Either ConnectError ()
initialResult of
    Left ConnectError
err -> do
      STM () -> IO ()
forall a. STM a -> IO a
atomically (TMVar () -> STM ()
forall a. TMVar a -> STM a
readTMVar TMVar ()
engineDone)
      ConnectionAPI -> Conn -> IO ()
Connection.close ConnectionAPI
connectionApi Conn
conn
      CallbackWorker -> IO ()
stopCallbackWorker CallbackWorker
callbackWorker
      ThreadId
current <- IO ThreadId
myThreadId
      ThreadId -> IO ()
stopAuxiliaryThreads ThreadId
current
      Either ConnectError Client -> IO (Either ConnectError Client)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (ConnectError -> Either ConnectError Client
forall a b. a -> Either a b
Left ConnectError
err)
    Right () -> Either ConnectError Client -> IO (Either ConnectError Client)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Client -> Either ConnectError Client
forall a b. b -> Either a b
Right Client
client)

type ConfigOption = CallOption ClientOptions

withConnectName :: BS.ByteString -> ConfigOption
withConnectName :: SID -> ConfigOption
withConnectName SID
name ClientOptions
config =
  ClientOptions
config
    { optionConnectConfig =
        (optionConnectConfig config) { Connect.name = Just name }
    }

withEcho :: Bool -> ConfigOption
withEcho :: Bool -> ConfigOption
withEcho Bool
enabled ClientOptions
config =
  ClientOptions
config
    { optionConnectConfig =
        (optionConnectConfig config) { Connect.echo = Just enabled }
    }

withAuthToken :: AuthTokenData -> ConfigOption
withAuthToken :: SID -> ConfigOption
withAuthToken SID
token = Auth -> ConfigOption
addAuth (SID -> Auth
AuthToken.auth SID
token)

-- | Fetch a token for every connection and reconnection attempt.
withAuthTokenHandler :: AuthTokenHandler -> ConfigOption
withAuthTokenHandler :: AuthTokenHandler -> ConfigOption
withAuthTokenHandler AuthTokenHandler
handler = Auth -> ConfigOption
addAuth (AuthTokenHandler -> Auth
AuthToken.authHandler AuthTokenHandler
handler)

withUserPass :: UserPassData -> ConfigOption
withUserPass :: UserPassData -> ConfigOption
withUserPass UserPassData
userPass = Auth -> ConfigOption
addAuth (UserPassData -> Auth
AuthUserPass.auth UserPassData
userPass)

-- | Fetch a username and password for every connection and reconnection attempt.
withUserPassHandler :: UserPassHandler -> ConfigOption
withUserPassHandler :: UserPassHandler -> ConfigOption
withUserPassHandler UserPassHandler
handler = Auth -> ConfigOption
addAuth (UserPassHandler -> Auth
AuthUserPass.authHandler UserPassHandler
handler)

withNKey :: NKeyData -> ConfigOption
withNKey :: SID -> ConfigOption
withNKey SID
seed = Auth -> ConfigOption
addAuth (SID -> Auth
AuthNKey.auth SID
seed)

-- | Authenticate with a public NKey and a handler that signs the server nonce.
-- The handler returns the raw 64-byte Ed25519 signature; natskell performs the
-- protocol's base64url encoding.
withNKeyHandler :: NKeyPublicKey -> SignatureHandler -> ConfigOption
withNKeyHandler :: SID -> SignatureHandler -> ConfigOption
withNKeyHandler SID
publicKey SignatureHandler
handler = Auth -> ConfigOption
addAuth (SID -> SignatureHandler -> Auth
AuthNKey.authHandler SID
publicKey SignatureHandler
handler)

withJWT :: JWTTokenData -> ConfigOption
withJWT :: SID -> ConfigOption
withJWT SID
creds = Auth -> ConfigOption
addAuth (SID -> Auth
AuthJwt.auth SID
creds)

-- | Fetch a user JWT and sign the server nonce for every connection attempt.
withJWTHandlers :: JWTHandler -> SignatureHandler -> ConfigOption
withJWTHandlers :: AuthTokenHandler -> SignatureHandler -> ConfigOption
withJWTHandlers AuthTokenHandler
jwtHandler SignatureHandler
signatureHandler =
  Auth -> ConfigOption
addAuth (AuthTokenHandler -> SignatureHandler -> Auth
AuthJwt.authHandlers AuthTokenHandler
jwtHandler SignatureHandler
signatureHandler)

addAuth :: Auth -> ConfigOption
addAuth :: Auth -> ConfigOption
addAuth Auth
auth ClientOptions
config =
  ClientOptions
config { optionAuth = mergeAuth (optionAuth config) auth }

-- | Require TLS using the operating system trust store.
withTLS :: ConfigOption
withTLS :: ConfigOption
withTLS = (TLSConfig -> TLSConfig) -> ConfigOption
modifyTLSConfig TLSConfig -> TLSConfig
forall a. a -> a
id

withTLSCert :: TLSCertData -> ConfigOption
withTLSCert :: UserPassData -> ConfigOption
withTLSCert UserPassData
cert =
  (TLSConfig -> TLSConfig) -> ConfigOption
modifyTLSConfig ((TLSConfig -> TLSConfig) -> ConfigOption)
-> (TLSConfig -> TLSConfig) -> ConfigOption
forall a b. (a -> b) -> a -> b
$ \TLSConfig
tls -> TLSConfig
tls { tlsClientCertificate = Just cert }

-- | Trust a PEM-encoded root certificate for this client. Once configured,
-- these roots replace the operating-system trust store for the connection.
withTLSRootCA :: BS.ByteString -> ConfigOption
withTLSRootCA :: SID -> ConfigOption
withTLSRootCA SID
root =
  (TLSConfig -> TLSConfig) -> ConfigOption
modifyTLSConfig ((TLSConfig -> TLSConfig) -> ConfigOption)
-> (TLSConfig -> TLSConfig) -> ConfigOption
forall a b. (a -> b) -> a -> b
$ \TLSConfig
tls ->
    TLSConfig
tls { tlsRootCertificates = tlsRootCertificates tls ++ [root] }

-- | Override the host name used for certificate verification and SNI.
withTLSServerName :: String -> ConfigOption
withTLSServerName :: [Char] -> ConfigOption
withTLSServerName [Char]
serverName =
  (TLSConfig -> TLSConfig) -> ConfigOption
modifyTLSConfig ((TLSConfig -> TLSConfig) -> ConfigOption)
-> (TLSConfig -> TLSConfig) -> ConfigOption
forall a b. (a -> b) -> a -> b
$ \TLSConfig
tls -> TLSConfig
tls { tlsServerName = Just serverName }

-- | Disable server certificate verification. This is unsafe and should only
-- be used when the peer is trusted by some mechanism outside TLS.
withTLSInsecure :: ConfigOption
withTLSInsecure :: ConfigOption
withTLSInsecure =
  (TLSConfig -> TLSConfig) -> ConfigOption
modifyTLSConfig ((TLSConfig -> TLSConfig) -> ConfigOption)
-> (TLSConfig -> TLSConfig) -> ConfigOption
forall a b. (a -> b) -> a -> b
$ \TLSConfig
tls -> TLSConfig
tls { tlsInsecure = True }

-- | Fetch complete mutual-TLS configuration before every TLS handshake.
--
-- This is intended for renewable identities such as SPIFFE SVIDs. It replaces
-- static TLS options; applying a static TLS option afterwards replaces source.
withTLSConfigSource :: TLSConfigSource -> ConfigOption
withTLSConfigSource :: TLSConfigSource -> ConfigOption
withTLSConfigSource TLSConfigSource
source ClientOptions
config =
  ClientOptions
config
    { optionTlsConfig = Nothing
    , optionTlsConfigSource = Just source
    }

modifyTLSConfig :: (TLSConfig -> TLSConfig) -> ConfigOption
modifyTLSConfig :: (TLSConfig -> TLSConfig) -> ConfigOption
modifyTLSConfig TLSConfig -> TLSConfig
update ClientOptions
config =
  ClientOptions
config
    { optionTlsConfig =
        Just (update (fromMaybe defaultTLSConfig (optionTlsConfig config)))
    , optionTlsConfigSource = Nothing
    }

withMinimumLogLevel :: LogLevel -> ConfigOption
withMinimumLogLevel :: LogLevel -> ConfigOption
withMinimumLogLevel LogLevel
minimumLogLevel ClientOptions
config =
  ClientOptions
config
    { optionLoggerConfig =
        (optionLoggerConfig config) { minLogLevel = minimumLogLevel }
    }

withLogAction :: (LogEntry -> IO ()) -> ConfigOption
withLogAction :: (LogEntry -> IO ()) -> ConfigOption
withLogAction LogEntry -> IO ()
logAction ClientOptions
config =
  ClientOptions
config
    { optionLoggerConfig =
        (optionLoggerConfig config) { logFn = logAction }
    }

withConnectionAttempts :: Int -> ConfigOption
withConnectionAttempts :: Int -> ConfigOption
withConnectionAttempts Int
attempts ClientOptions
config =
  ClientOptions
config { optionConnectionAttempts = max 1 attempts }

-- | Set the total time budget for TCP acquisition, INFO, TLS, CONNECT, PONG,
-- and reconnect resubscription/readiness on each server attempt. Values below
-- one microsecond are clamped to one.
withConnectTimeoutMicros :: Int -> ConfigOption
withConnectTimeoutMicros :: Int -> ConfigOption
withConnectTimeoutMicros Int
timeoutMicros ClientOptions
config =
  ClientOptions
config { optionConnectTimeoutMicros = max 1 timeoutMicros }

withCallbackConcurrency :: Int -> ConfigOption
withCallbackConcurrency :: Int -> ConfigOption
withCallbackConcurrency Int
concurrency ClientOptions
config =
  ClientOptions
config { optionCallbackConcurrency = concurrency }

-- | Set the largest encoded message body this client accepts. For messages
-- with headers, the encoded header block and payload both count toward the
-- limit. Outbound messages are also constrained by the server's max_payload.
withMessageLimit :: Int -> ConfigOption
withMessageLimit :: Int -> ConfigOption
withMessageLimit Int
limit ClientOptions
config =
  ClientOptions
config { optionMessageLimit = max 1 limit }

-- | Bound callback deliveries pending across the entire client. The first
-- limit is a message count and the second is encoded message bytes.
withPendingDeliveryLimits :: Int -> Int -> ConfigOption
withPendingDeliveryLimits :: Int -> Int -> ConfigOption
withPendingDeliveryLimits Int
maximumMessages Int
maximumBytes ClientOptions
config =
  ClientOptions
config
    { optionPendingLimits =
        PendingLimits
          { pendingMessageLimit = max 1 maximumMessages
          , pendingByteLimit = max 1 maximumBytes
          }
    }

-- | Receive asynchronous client errors such as slow-consumer notifications.
-- The handler runs on the callback worker pool, never the socket reader.
withErrorHandler :: (NatsError -> IO ()) -> ConfigOption
withErrorHandler :: (NatsError -> IO ()) -> ConfigOption
withErrorHandler NatsError -> IO ()
handler ClientOptions
config =
  ClientOptions
config { optionErrorHandler = handler }

-- | Receive protocol @-ERR@ values on a bounded, dedicated serial worker.
-- Protocol processing, including PONG handling, does not wait for the handler.
withServerErrorHandler :: (ServerError -> IO ()) -> ConfigOption
withServerErrorHandler :: (ServerError -> IO ()) -> ConfigOption
withServerErrorHandler ServerError -> IO ()
handler ClientOptions
config =
  ClientOptions
config { optionServerErrorHandler = handler }

-- | Receive completed disconnect, reconnect, and close transitions on the
-- same serial callback worker as server errors. Lifecycle delivery is reserved
-- independently from the bounded server-error backlog.
withConnectionEventHandler :: (ConnectionEvent -> IO ()) -> ConfigOption
withConnectionEventHandler :: (ConnectionEvent -> IO ()) -> ConfigOption
withConnectionEventHandler ConnectionEvent -> IO ()
handler ClientOptions
config =
  ClientOptions
config { optionConnectionEventHandler = handler }

-- | Compatibility alias for 'withMessageLimit'.
withBufferLimit :: Int -> ConfigOption
withBufferLimit :: Int -> ConfigOption
withBufferLimit = Int -> ConfigOption
withMessageLimit

{-# DEPRECATED withBufferLimit "Use withMessageLimit instead." #-}

withExitAction :: (ClientExitReason -> IO ()) -> ConfigOption
withExitAction :: (ClientExitReason -> IO ()) -> ConfigOption
withExitAction ClientExitReason -> IO ()
action ClientOptions
config = ClientOptions
config { optionExitAction = action }

defaultConnect :: Connect.Connect
defaultConnect :: Connect
defaultConnect =
  Connect.Connect
    { verbose :: Bool
Connect.verbose = Bool
False
    , pedantic :: Bool
Connect.pedantic = Bool
True
    , tls_required :: Bool
Connect.tls_required = Bool
False
    , auth_token :: Maybe SID
Connect.auth_token = Maybe SID
forall a. Maybe a
Nothing
    , user :: Maybe SID
Connect.user = Maybe SID
forall a. Maybe a
Nothing
    , pass :: Maybe SID
Connect.pass = Maybe SID
forall a. Maybe a
Nothing
    , name :: Maybe SID
Connect.name = Maybe SID
forall a. Maybe a
Nothing
    , lang :: SID
Connect.lang = SID
"haskell"
    , version :: SID
Connect.version = [Char] -> SID
BC.pack (Version -> [Char]
showVersion Version
Package.version)
    , protocol :: Maybe Int
Connect.protocol = Maybe Int
forall a. Maybe a
Nothing
    , echo :: Maybe Bool
Connect.echo = Bool -> Maybe Bool
forall a. a -> Maybe a
Just Bool
True
    , sig :: Maybe SID
Connect.sig = Maybe SID
forall a. Maybe a
Nothing
    , jwt :: Maybe SID
Connect.jwt = Maybe SID
forall a. Maybe a
Nothing
    , nkey :: Maybe SID
Connect.nkey = Maybe SID
forall a. Maybe a
Nothing
    , no_responders :: Maybe Bool
Connect.no_responders = Bool -> Maybe Bool
forall a. a -> Maybe a
Just Bool
True
    , headers :: Maybe Bool
Connect.headers = Bool -> Maybe Bool
forall a. a -> Maybe a
Just Bool
True
    }

defaultSubscribeConfig :: SubscribeConfig
defaultSubscribeConfig :: SubscribeConfig
defaultSubscribeConfig = Maybe NominalDiffTime -> Maybe SID -> SubscribeConfig
SubscribeConfig Maybe NominalDiffTime
forall a. Maybe a
Nothing Maybe SID
forall a. Maybe a
Nothing

defaultRequestConfig :: RequestConfig
defaultRequestConfig :: RequestConfig
defaultRequestConfig = NominalDiffTime -> Maybe Headers -> RequestConfig
RequestConfig NominalDiffTime
2 Maybe Headers
forall a. Maybe a
Nothing

defaultPingConfig :: PingConfig
defaultPingConfig :: PingConfig
defaultPingConfig = PingConfig
PingConfig

defaultFlushConfig :: FlushConfig
defaultFlushConfig :: FlushConfig
defaultFlushConfig = FlushConfig
FlushConfig

defaultRoundTripTimeout :: NominalDiffTime
defaultRoundTripTimeout :: NominalDiffTime
defaultRoundTripTimeout = NominalDiffTime
2

logStaticConfiguration :: ClientState -> ClientOptions -> IO ()
logStaticConfiguration :: ClientState -> ClientOptions -> IO ()
logStaticConfiguration ClientState
client ClientOptions
options =
  ClientState -> AppM () -> IO ()
forall a. ClientState -> AppM a -> IO a
runClient ClientState
client (AppM () -> IO ()) -> AppM () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
    case Auth -> [[Char]]
authMethods (ClientOptions -> Auth
optionAuth ClientOptions
options) of
      [] -> LogLevel -> [Char] -> AppM ()
forall (m :: * -> *). MonadLogger m => LogLevel -> [Char] -> m ()
logMessage LogLevel
Info [Char]
"no authentication method provided"
      [[Char]]
methods -> [[Char]] -> ([Char] -> AppM ()) -> AppM ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [[Char]]
methods (([Char] -> AppM ()) -> AppM ()) -> ([Char] -> AppM ()) -> AppM ()
forall a b. (a -> b) -> a -> b
$ \[Char]
method ->
        LogLevel -> [Char] -> AppM ()
forall (m :: * -> *). MonadLogger m => LogLevel -> [Char] -> m ()
logMessage LogLevel
Info ([Char]
"using " [Char] -> ShowS
forall a. [a] -> [a] -> [a]
++ [Char]
method)
    case (ClientOptions -> Maybe TLSConfig
optionTlsConfig ClientOptions
options, ClientOptions -> Maybe TLSConfigSource
optionTlsConfigSource ClientOptions
options) of
      (Maybe TLSConfig
Nothing, Maybe TLSConfigSource
Nothing) -> () -> AppM ()
forall a. a -> AppM a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
      (Maybe TLSConfig
Nothing, Just TLSConfigSource
_) -> LogLevel -> [Char] -> AppM ()
forall (m :: * -> *). MonadLogger m => LogLevel -> [Char] -> m ()
logMessage LogLevel
Info [Char]
"using renewable tls configuration"
      (Just TLSConfig
tls, Maybe TLSConfigSource
_) -> do
        LogLevel -> [Char] -> AppM ()
forall (m :: * -> *). MonadLogger m => LogLevel -> [Char] -> m ()
logMessage LogLevel
Info [Char]
"using tls"
        Bool -> AppM () -> AppM ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (TLSConfig -> Bool
tlsInsecure TLSConfig
tls) (AppM () -> AppM ()) -> AppM () -> AppM ()
forall a b. (a -> b) -> a -> b
$
          LogLevel -> [Char] -> AppM ()
forall (m :: * -> *). MonadLogger m => LogLevel -> [Char] -> m ()
logMessage LogLevel
Warn [Char]
"tls certificate verification is disabled"

handleCallbackError :: ClientState -> SomeException -> IO ()
handleCallbackError :: ClientState -> SomeException -> IO ()
handleCallbackError ClientState
client SomeException
_ =
  ClientState -> AppM () -> IO ()
forall a. ClientState -> AppM a -> IO a
runClient ClientState
client (AppM () -> IO ()) -> AppM () -> IO ()
forall a b. (a -> b) -> a -> b
$
    LogLevel -> [Char] -> AppM ()
forall (m :: * -> *). MonadLogger m => LogLevel -> [Char] -> m ()
logMessage LogLevel
Error [Char]
"callback failed"

handleSlowConsumer :: ClientState -> (NatsError -> IO ()) -> IO ()
handleSlowConsumer :: ClientState -> (NatsError -> IO ()) -> IO ()
handleSlowConsumer ClientState
client NatsError -> IO ()
handler = do
  ClientState -> AppM () -> IO ()
forall a. ClientState -> AppM a -> IO a
runClient ClientState
client (AppM () -> IO ()) -> AppM () -> IO ()
forall a b. (a -> b) -> a -> b
$
    LogLevel -> [Char] -> AppM ()
forall (m :: * -> *). MonadLogger m => LogLevel -> [Char] -> m ()
logMessage LogLevel
Error [Char]
"slow consumer: global pending delivery limit reached"
  NatsError -> IO ()
handler NatsError
NatsSlowConsumer

shouldStopExpiryWorker :: ClientState -> SubscriptionStore -> IO Bool
shouldStopExpiryWorker :: ClientState -> SubscriptionStore -> IO Bool
shouldStopExpiryWorker ClientState
client SubscriptionStore
store = do
  ConnectionState
status <- ClientState -> IO ConnectionState
readStatus ClientState
client
  Bool
tracked <- SubscriptionStore -> IO Bool
hasTrackedExpiries SubscriptionStore
store
  Bool -> IO Bool
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Bool -> IO Bool) -> Bool -> IO Bool
forall a b. (a -> b) -> a -> b
$
    case ConnectionState
status of
      ConnectionClosed ClientExitReason
_ -> Bool -> Bool
not Bool
tracked
      ConnectionState
_                  -> Bool
False

toMessage :: Msg.Msg -> Message
toMessage :: Msg -> Message
toMessage Msg
msg =
  Message
    { subject :: SID
subject = Msg -> SID
Msg.subject Msg
msg
    , sid :: SID
sid = Msg -> SID
Msg.sid Msg
msg
    , replyTo :: Maybe SID
replyTo = Msg -> Maybe SID
Msg.replyTo Msg
msg
    , payload :: SID
payload = SID -> Maybe SID -> SID
forall a. a -> Maybe a -> a
fromMaybe SID
BS.empty (Msg -> Maybe SID
Msg.payload Msg
msg)
    , headers :: Maybe Headers
headers = Msg -> Maybe Headers
Msg.headers Msg
msg
    }

publishClient
  :: ClientState
  -> (NatsError -> IO ())
  -> Msg.Subject
  -> Msg.Payload
  -> PublishConfig
  -> IO (Either NatsError ())
publishClient :: ClientState
-> (NatsError -> IO ())
-> SID
-> SID
-> PublishConfig
-> IO (Either NatsError ())
publishClient ClientState
client NatsError -> IO ()
errorHandler SID
subject SID
messagePayload PublishConfig
cfg = do
  ClientState -> AppM () -> IO ()
forall a. ClientState -> AppM a -> IO a
runClient ClientState
client (AppM () -> IO ()) -> AppM () -> IO ()
forall a b. (a -> b) -> a -> b
$
    LogLevel -> [Char] -> AppM ()
forall (m :: * -> *). MonadLogger m => LogLevel -> [Char] -> m ()
logMessage LogLevel
Debug ([Char]
"publishing to subject: " [Char] -> ShowS
forall a. [a] -> [a] -> [a]
++ SID -> [Char]
forall a. Show a => a -> [Char]
show SID
subject)
  let publishMessage :: Pub
publishMessage =
        Pub.Pub
          { subject :: SID
Pub.subject = SID
subject
          , payload :: Maybe SID
Pub.payload =
              if SID -> Bool
BS.null SID
messagePayload then Maybe SID
forall a. Maybe a
Nothing else SID -> Maybe SID
forall a. a -> Maybe a
Just SID
messagePayload
          , replyTo :: Maybe SID
Pub.replyTo = PublishConfig -> Maybe SID
publishReplyTo PublishConfig
cfg
          , headers :: Maybe Headers
Pub.headers = PublishConfig -> Maybe Headers
publishHeaders PublishConfig
cfg
          }
      actualSize :: Int
actualSize = Pub -> Int
Pub.messageSize Pub
publishMessage
      queuedPublish :: QueueItem
queuedPublish =
        Int -> (Int -> IO ()) -> QueueItem -> QueueItem
QueuePayloadBound
          Int
actualSize
          (NatsError -> IO ()
errorHandler (NatsError -> IO ()) -> (Int -> NatsError) -> Int -> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Int -> Int -> NatsError
NatsPayloadTooLarge Int
actualSize)
          (Pub -> QueueItem
forall m. Transformer m => m -> QueueItem
QueueItem Pub
publishMessage)
  Either NatsError ()
validation <- ClientState -> Pub -> IO (Either NatsError ())
validatePublish ClientState
client Pub
publishMessage
  case Either NatsError ()
validation of
    Left NatsError
err -> Either NatsError () -> IO (Either NatsError ())
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (NatsError -> Either NatsError ()
forall a b. a -> Either a b
Left NatsError
err)
    Right () -> do
      PublishEnqueueResult
enqueueResult <- ClientState -> Int -> QueueItem -> IO PublishEnqueueResult
enqueuePublishOnConnected ClientState
client Int
actualSize QueueItem
queuedPublish
      ClientState
-> Int -> PublishEnqueueResult -> IO (Either NatsError ())
publishEnqueueResult ClientState
client Int
actualSize PublishEnqueueResult
enqueueResult

validatePublish :: ClientState -> Pub.Pub -> IO (Either NatsError ())
validatePublish :: ClientState -> Pub -> IO (Either NatsError ())
validatePublish ClientState
client Pub
publishMessage =
  case Pub -> Either SID ()
forall a. Validator a => a -> Either SID ()
validate Pub
publishMessage of
    Left SID
reason -> do
      ClientState -> AppM () -> IO ()
forall a. ClientState -> AppM a -> IO a
runClient ClientState
client (AppM () -> IO ()) -> AppM () -> IO ()
forall a b. (a -> b) -> a -> b
$
        LogLevel -> [Char] -> AppM ()
forall (m :: * -> *). MonadLogger m => LogLevel -> [Char] -> m ()
logMessage LogLevel
Error ([Char]
"rejecting invalid publish: " [Char] -> ShowS
forall a. [a] -> [a] -> [a]
++ SID -> [Char]
BC.unpack SID
reason)
      Either NatsError () -> IO (Either NatsError ())
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (NatsError -> Either NatsError ()
forall a b. a -> Either a b
Left (SID -> NatsError
NatsValidationError SID
reason))
    Right () -> do
      let actualSize :: Int
actualSize = Pub -> Int
Pub.messageSize Pub
publishMessage
          clientMaximum :: Int
clientMaximum = ClientConfig -> Int
messageLimit (ClientState -> ClientConfig
config ClientState
client)
      if Int
actualSize Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
> Int
clientMaximum
        then ClientState -> Int -> Int -> IO (Either NatsError ())
forall a. ClientState -> Int -> Int -> IO (Either NatsError a)
rejectPayloadTooLarge ClientState
client Int
actualSize Int
clientMaximum
        else Either NatsError () -> IO (Either NatsError ())
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (() -> Either NatsError ()
forall a b. b -> Either a b
Right ())

publishEnqueueResult
  :: ClientState
  -> Int
  -> PublishEnqueueResult
  -> IO (Either NatsError ())
publishEnqueueResult :: ClientState
-> Int -> PublishEnqueueResult -> IO (Either NatsError ())
publishEnqueueResult ClientState
client Int
actualSize PublishEnqueueResult
enqueueResult =
  case PublishEnqueueResult
enqueueResult of
    PublishEnqueued Int
_ -> Either NatsError () -> IO (Either NatsError ())
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (() -> Either NatsError ()
forall a b. b -> Either a b
Right ())
    PublishTooLarge Int
maximumSize ->
      ClientState -> Int -> Int -> IO (Either NatsError ())
forall a. ClientState -> Int -> Int -> IO (Either NatsError a)
rejectPayloadTooLarge ClientState
client Int
actualSize Int
maximumSize
    PublishConnectionClosed ClientExitReason
reason ->
      Either NatsError () -> IO (Either NatsError ())
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (NatsError -> Either NatsError ()
forall a b. a -> Either a b
Left (ClientExitReason -> NatsError
NatsConnectionClosed ClientExitReason
reason))

rejectPayloadTooLarge :: ClientState -> Int -> Int -> IO (Either NatsError a)
rejectPayloadTooLarge :: forall a. ClientState -> Int -> Int -> IO (Either NatsError a)
rejectPayloadTooLarge ClientState
client Int
actualSize Int
maximumSize = do
  ClientState -> AppM () -> IO ()
forall a. ClientState -> AppM a -> IO a
runClient ClientState
client (AppM () -> IO ()) -> AppM () -> IO ()
forall a b. (a -> b) -> a -> b
$
    LogLevel -> [Char] -> AppM ()
forall (m :: * -> *). MonadLogger m => LogLevel -> [Char] -> m ()
logMessage LogLevel
Error
      ( [Char]
"rejecting publish: message size "
          [Char] -> ShowS
forall a. [a] -> [a] -> [a]
++ Int -> [Char]
forall a. Show a => a -> [Char]
show Int
actualSize
          [Char] -> ShowS
forall a. [a] -> [a] -> [a]
++ [Char]
" exceeds effective limit "
          [Char] -> ShowS
forall a. [a] -> [a] -> [a]
++ Int -> [Char]
forall a. Show a => a -> [Char]
show Int
maximumSize
      )
  Either NatsError a -> IO (Either NatsError a)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (NatsError -> Either NatsError a
forall a b. a -> Either a b
Left (Int -> Int -> NatsError
NatsPayloadTooLarge Int
actualSize Int
maximumSize))

subscribeClient
  :: ClientState
  -> SubscriptionStore
  -> SubscriptionKind
  -> Msg.Subject
  -> SubscribeConfig
  -> (Message -> IO ())
  -> IO (Either NatsError Subscription)
subscribeClient :: ClientState
-> SubscriptionStore
-> SubscriptionKind
-> SID
-> SubscribeConfig
-> (Message -> IO ())
-> IO (Either NatsError Subscription)
subscribeClient ClientState
client SubscriptionStore
store SubscriptionKind
subscriptionKind SID
subject SubscribeConfig
cfg Message -> IO ()
callback =
  ClientState
-> SubscriptionStore
-> SubscriptionKind
-> SID
-> SubscribeConfig
-> (Maybe Msg -> IO ())
-> IO (Either NatsError Subscription)
subscribeRawClient ClientState
client SubscriptionStore
store SubscriptionKind
subscriptionKind SID
subject SubscribeConfig
cfg (IO () -> (Msg -> IO ()) -> Maybe Msg -> IO ()
forall b a. b -> (a -> b) -> Maybe a -> b
maybe (() -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()) (Message -> IO ()
callback (Message -> IO ()) -> (Msg -> Message) -> Msg -> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Msg -> Message
toMessage))

subscribeRawClient
  :: ClientState
  -> SubscriptionStore
  -> SubscriptionKind
  -> Msg.Subject
  -> SubscribeConfig
  -> (Maybe Msg.Msg -> IO ())
  -> IO (Either NatsError Subscription)
subscribeRawClient :: ClientState
-> SubscriptionStore
-> SubscriptionKind
-> SID
-> SubscribeConfig
-> (Maybe Msg -> IO ())
-> IO (Either NatsError Subscription)
subscribeRawClient ClientState
client SubscriptionStore
store SubscriptionKind
subscriptionKind SID
subject SubscribeConfig
cfg Maybe Msg -> IO ()
callback = do
  ClientState
-> SubscriptionStore
-> SubscriptionKind
-> SID
-> SubscribeConfig
-> (Maybe Msg -> IO ())
-> IO ()
-> IO (Either NatsError Subscription)
subscribeRawClientWithOverflow
    ClientState
client
    SubscriptionStore
store
    SubscriptionKind
subscriptionKind
    SID
subject
    SubscribeConfig
cfg
    Maybe Msg -> IO ()
callback
    (() -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ())

subscribeRawClientWithOverflow
  :: ClientState
  -> SubscriptionStore
  -> SubscriptionKind
  -> Msg.Subject
  -> SubscribeConfig
  -> (Maybe Msg.Msg -> IO ())
  -> IO ()
  -> IO (Either NatsError Subscription)
subscribeRawClientWithOverflow :: ClientState
-> SubscriptionStore
-> SubscriptionKind
-> SID
-> SubscribeConfig
-> (Maybe Msg -> IO ())
-> IO ()
-> IO (Either NatsError Subscription)
subscribeRawClientWithOverflow ClientState
client SubscriptionStore
store SubscriptionKind
subscriptionKind SID
subject SubscribeConfig
cfg Maybe Msg -> IO ()
callback IO ()
onDropped =
  ((forall a. IO a -> IO a) -> IO (Either NatsError Subscription))
-> IO (Either NatsError Subscription)
forall b. ((forall a. IO a -> IO a) -> IO b) -> IO b
mask (((forall a. IO a -> IO a) -> IO (Either NatsError Subscription))
 -> IO (Either NatsError Subscription))
-> ((forall a. IO a -> IO a) -> IO (Either NatsError Subscription))
-> IO (Either NatsError Subscription)
forall a b. (a -> b) -> a -> b
$ \forall a. IO a -> IO a
restore -> do
    ClientState -> AppM () -> IO ()
forall a. ClientState -> AppM a -> IO a
runClient ClientState
client (AppM () -> IO ()) -> AppM () -> IO ()
forall a b. (a -> b) -> a -> b
$
      LogLevel -> [Char] -> AppM ()
forall (m :: * -> *). MonadLogger m => LogLevel -> [Char] -> m ()
logMessage LogLevel
Debug ([Char]
"subscribing to subject: " [Char] -> ShowS
forall a. [a] -> [a] -> [a]
++ SID -> [Char]
forall a. Show a => a -> [Char]
show SID
subject)
    let queueGroup :: Maybe SID
queueGroup = SubscribeConfig -> Maybe SID
subscribeQueueGroup SubscribeConfig
cfg
    SID
sid <- ClientState -> IO SID
nextSid ClientState
client
    TMVar Int
registeredGeneration <- IO (TMVar Int)
forall a. IO (TMVar a)
newEmptyTMVarIO
    TMVar Int
committedGeneration <- IO (TMVar Int)
forall a. IO (TMVar a)
newEmptyTMVarIO
    let subscriptionMessage :: Sub
subscriptionMessage =
          Sub.Sub
            { subject :: SID
Sub.subject = SID
subject
            , queueGroup :: Maybe SID
Sub.queueGroup = Maybe SID
queueGroup
            , sid :: SID
Sub.sid = SID
sid
            }
        meta :: SubscriptionMeta
meta = SID -> Maybe SID -> SubscriptionKind -> SubscriptionMeta
SubscriptionMeta SID
subject Maybe SID
queueGroup SubscriptionKind
subscriptionKind
        subscription :: Subscription
subscription = SID -> Subscription
Subscription SID
sid
        cleanup :: IO ()
cleanup =
          ClientState
-> SubscriptionStore
-> TMVar Int
-> TMVar Int
-> Subscription
-> IO ()
cleanupResumableSubscription
            ClientState
client
            SubscriptionStore
store
            TMVar Int
registeredGeneration
            TMVar Int
committedGeneration
            Subscription
subscription
        deliver :: Maybe Msg -> IO ()
deliver Maybe Msg
maybeMessage = do
          case Maybe Msg
maybeMessage of
            Maybe Msg
Nothing -> IO ()
cleanup
            Just Msg
_  -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
          Maybe Msg -> IO ()
callback Maybe Msg
maybeMessage
        commands :: [QueueItem]
commands =
          Sub -> QueueItem
forall m. Transformer m => m -> QueueItem
QueueItem Sub
subscriptionMessage QueueItem -> [QueueItem] -> [QueueItem]
forall a. a -> [a] -> [a]
:
            [ Unsub -> QueueItem
forall m. Transformer m => m -> QueueItem
QueueItem
                Unsub.Unsub
                  { sid :: SID
Unsub.sid = SID
sid
                  , maxMsg :: Maybe Int
Unsub.maxMsg = Int -> Maybe Int
forall a. a -> Maybe a
Just Int
1
                  }
            | SubscriptionMeta -> Bool
isOneShotSubscription SubscriptionMeta
meta
            ]
        queuedCommands :: QueueItem
queuedCommands = QueueItem -> QueueItem
QueueConnectionScoped ([QueueItem] -> QueueItem
QueueBatch [QueueItem]
commands)
    case Sub -> Either SID ()
forall a. Validator a => a -> Either SID ()
validate Sub
subscriptionMessage of
      Left SID
reason -> do
        ClientState -> AppM () -> IO ()
forall a. ClientState -> AppM a -> IO a
runClient ClientState
client (AppM () -> IO ()) -> AppM () -> IO ()
forall a b. (a -> b) -> a -> b
$
          LogLevel -> [Char] -> AppM ()
forall (m :: * -> *). MonadLogger m => LogLevel -> [Char] -> m ()
logMessage LogLevel
Error ([Char]
"rejecting invalid subscription: " [Char] -> ShowS
forall a. [a] -> [a] -> [a]
++ SID -> [Char]
BC.unpack SID
reason)
        Either NatsError Subscription -> IO (Either NatsError Subscription)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (NatsError -> Either NatsError Subscription
forall a b. a -> Either a b
Left (SID -> NatsError
NatsValidationError SID
reason))
      Right () -> do
        let waitForServerRegistration :: IO (Either NatsError Subscription)
waitForServerRegistration = do
              Either NatsError ()
statusResult <- ClientState -> IO (Either NatsError ())
runningResult ClientState
client
              case Either NatsError ()
statusResult of
                Left NatsError
err -> IO ()
cleanup IO ()
-> IO (Either NatsError Subscription)
-> IO (Either NatsError Subscription)
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> Either NatsError Subscription -> IO (Either NatsError Subscription)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (NatsError -> Either NatsError Subscription
forall a b. a -> Either a b
Left NatsError
err)
                Right () -> Either NatsError Subscription -> IO (Either NatsError Subscription)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Subscription -> Either NatsError Subscription
forall a b. b -> Either a b
Right Subscription
subscription)
            acquireGate :: IO (Either NatsError SubscriptionQueueResult)
acquireGate = do
              Either NatsError ()
statusResult <- ClientState -> IO (Either NatsError ())
runningResult ClientState
client
              case Either NatsError ()
statusResult of
                Left NatsError
err -> Either NatsError SubscriptionQueueResult
-> IO (Either NatsError SubscriptionQueueResult)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (NatsError -> Either NatsError SubscriptionQueueResult
forall a b. a -> Either a b
Left NatsError
err)
                Right () -> do
                  Maybe (Either NatsError SubscriptionQueueResult)
gated <- ClientState
-> IO (Maybe (Either NatsError SubscriptionQueueResult))
-> IO (Maybe (Either NatsError SubscriptionQueueResult))
forall a. ClientState -> IO a -> IO a
withSubscriptionGate ClientState
client (IO (Maybe (Either NatsError SubscriptionQueueResult))
 -> IO (Maybe (Either NatsError SubscriptionQueueResult)))
-> IO (Maybe (Either NatsError SubscriptionQueueResult))
-> IO (Maybe (Either NatsError SubscriptionQueueResult))
forall a b. (a -> b) -> a -> b
$ do
                    ConnectionState
status <- ClientState -> IO ConnectionState
readStatus ClientState
client
                    case ConnectionState
status of
                      ConnectionState
ConnectionConnected -> do
                        Int
generation <- ClientState -> IO Int
readConnectionGeneration ClientState
client
                        STM () -> IO ()
forall a. STM a -> IO a
atomically (STM Bool -> STM ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (TMVar Int -> Int -> STM Bool
forall a. TMVar a -> a -> STM Bool
tryPutTMVar TMVar Int
registeredGeneration Int
generation))
                        SubscriptionStore
-> SID
-> SubscriptionMeta
-> SubscribeConfig
-> (Maybe Msg -> IO ())
-> IO ()
-> IO ()
register SubscriptionStore
store SID
sid SubscriptionMeta
meta SubscribeConfig
cfg Maybe Msg -> IO ()
deliver IO ()
onDropped
                        Either ClientExitReason Int
enqueueResult <-
                          ClientState
-> Int
-> TMVar Int
-> QueueItem
-> IO (Either ClientExitReason Int)
enqueueOnGenerationTracked
                            ClientState
client
                            Int
generation
                            TMVar Int
committedGeneration
                            QueueItem
queuedCommands
                        case Either ClientExitReason Int
enqueueResult of
                          Right Int
_ ->
                            Maybe (Either NatsError SubscriptionQueueResult)
-> IO (Maybe (Either NatsError SubscriptionQueueResult))
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Either NatsError SubscriptionQueueResult
-> Maybe (Either NatsError SubscriptionQueueResult)
forall a. a -> Maybe a
Just (SubscriptionQueueResult -> Either NatsError SubscriptionQueueResult
forall a b. b -> Either a b
Right SubscriptionQueueResult
SubscriptionQueued))
                          Left ClientExitReason
_ -> do
                            ConnectionState
nextStatus <- ClientState -> IO ConnectionState
readStatus ClientState
client
                            case ConnectionState
nextStatus of
                              ConnectionClosing ClientExitReason
reason ->
                                Maybe (Either NatsError SubscriptionQueueResult)
-> IO (Maybe (Either NatsError SubscriptionQueueResult))
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Either NatsError SubscriptionQueueResult
-> Maybe (Either NatsError SubscriptionQueueResult)
forall a. a -> Maybe a
Just (NatsError -> Either NatsError SubscriptionQueueResult
forall a b. a -> Either a b
Left (ClientExitReason -> NatsError
NatsConnectionClosed ClientExitReason
reason)))
                              ConnectionClosed ClientExitReason
reason ->
                                Maybe (Either NatsError SubscriptionQueueResult)
-> IO (Maybe (Either NatsError SubscriptionQueueResult))
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Either NatsError SubscriptionQueueResult
-> Maybe (Either NatsError SubscriptionQueueResult)
forall a. a -> Maybe a
Just (NatsError -> Either NatsError SubscriptionQueueResult
forall a b. a -> Either a b
Left (ClientExitReason -> NatsError
NatsConnectionClosed ClientExitReason
reason)))
                              ConnectionState
_ ->
                                Maybe (Either NatsError SubscriptionQueueResult)
-> IO (Maybe (Either NatsError SubscriptionQueueResult))
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Either NatsError SubscriptionQueueResult
-> Maybe (Either NatsError SubscriptionQueueResult)
forall a. a -> Maybe a
Just (SubscriptionQueueResult -> Either NatsError SubscriptionQueueResult
forall a b. b -> Either a b
Right SubscriptionQueueResult
SubscriptionReconnect))
                      ConnectionState
_ -> Maybe (Either NatsError SubscriptionQueueResult)
-> IO (Maybe (Either NatsError SubscriptionQueueResult))
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Maybe (Either NatsError SubscriptionQueueResult)
forall a. Maybe a
Nothing
                  IO (Either NatsError SubscriptionQueueResult)
-> (Either NatsError SubscriptionQueueResult
    -> IO (Either NatsError SubscriptionQueueResult))
-> Maybe (Either NatsError SubscriptionQueueResult)
-> IO (Either NatsError SubscriptionQueueResult)
forall b a. b -> (a -> b) -> Maybe a -> b
maybe IO (Either NatsError SubscriptionQueueResult)
acquireGate Either NatsError SubscriptionQueueResult
-> IO (Either NatsError SubscriptionQueueResult)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Maybe (Either NatsError SubscriptionQueueResult)
gated
            subscribeOperation :: IO (Either NatsError Subscription)
subscribeOperation = do
              Either NatsError SubscriptionQueueResult
queueResult <- IO (Either NatsError SubscriptionQueueResult)
acquireGate
              case Either NatsError SubscriptionQueueResult
queueResult of
                Left NatsError
err                    -> IO ()
cleanup IO ()
-> IO (Either NatsError Subscription)
-> IO (Either NatsError Subscription)
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> Either NatsError Subscription -> IO (Either NatsError Subscription)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (NatsError -> Either NatsError Subscription
forall a b. a -> Either a b
Left NatsError
err)
                Right SubscriptionQueueResult
SubscriptionQueued -> do
                  ClientState -> AppM () -> IO ()
forall a. ClientState -> AppM a -> IO a
runClient ClientState
client (AppM () -> IO ()) -> AppM () -> IO ()
forall a b. (a -> b) -> a -> b
$
                    LogLevel -> [Char] -> AppM ()
forall (m :: * -> *). MonadLogger m => LogLevel -> [Char] -> m ()
logMessage LogLevel
Debug [Char]
"subscription commands queued"
                  Either NatsError Subscription -> IO (Either NatsError Subscription)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Subscription -> Either NatsError Subscription
forall a b. b -> Either a b
Right Subscription
subscription)
                Right SubscriptionQueueResult
SubscriptionReconnect -> IO (Either NatsError Subscription)
waitForServerRegistration
        IO (Either NatsError Subscription)
-> IO (Either NatsError Subscription)
forall a. IO a -> IO a
restore IO (Either NatsError Subscription)
subscribeOperation IO (Either NatsError Subscription)
-> IO () -> IO (Either NatsError Subscription)
forall a b. IO a -> IO b -> IO a
`onException` IO ()
cleanup

requestClient
  :: ClientState
  -> SubscriptionStore
  -> Msg.Subject
  -> Msg.Payload
  -> RequestConfig
  -> IO (Either NatsError Message)
requestClient :: ClientState
-> SubscriptionStore
-> SID
-> SID
-> RequestConfig
-> IO (Either NatsError Message)
requestClient ClientState
client SubscriptionStore
store SID
requestSubject SID
requestPayload RequestConfig
cfg = do
  Maybe (Either NatsError Message)
timedResult <-
    Int
-> IO (Either NatsError Message)
-> IO (Maybe (Either NatsError Message))
forall a. Int -> IO a -> IO (Maybe a)
timeout
      (NominalDiffTime -> Int
durationMicros (RequestConfig -> NominalDiffTime
requestTimeout RequestConfig
cfg))
      (ClientState
-> SubscriptionStore
-> SID
-> SID
-> RequestConfig
-> IO (Either NatsError Message)
requestBeforeDeadline ClientState
client SubscriptionStore
store SID
requestSubject SID
requestPayload RequestConfig
cfg)
  Either NatsError Message -> IO (Either NatsError Message)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Either NatsError Message
-> Maybe (Either NatsError Message) -> Either NatsError Message
forall a. a -> Maybe a -> a
fromMaybe (NatsError -> Either NatsError Message
forall a b. a -> Either a b
Left NatsError
NatsRequestTimedOut) Maybe (Either NatsError Message)
timedResult)

requestBeforeDeadline
  :: ClientState
  -> SubscriptionStore
  -> Msg.Subject
  -> Msg.Payload
  -> RequestConfig
  -> IO (Either NatsError Message)
requestBeforeDeadline :: ClientState
-> SubscriptionStore
-> SID
-> SID
-> RequestConfig
-> IO (Either NatsError Message)
requestBeforeDeadline ClientState
client SubscriptionStore
store SID
requestSubject SID
requestPayload RequestConfig
cfg =
  ((forall a. IO a -> IO a) -> IO (Either NatsError Message))
-> IO (Either NatsError Message)
forall b. ((forall a. IO a -> IO a) -> IO b) -> IO b
mask (((forall a. IO a -> IO a) -> IO (Either NatsError Message))
 -> IO (Either NatsError Message))
-> ((forall a. IO a -> IO a) -> IO (Either NatsError Message))
-> IO (Either NatsError Message)
forall a b. (a -> b) -> a -> b
$ \forall a. IO a -> IO a
restore -> do
    TMVar (Either NatsError Message)
response <- IO (TMVar (Either NatsError Message))
forall a. IO (TMVar a)
newEmptyTMVarIO
    TMVar ()
accepted <- IO (TMVar ())
forall a. IO (TMVar a)
newEmptyTMVarIO
    TMVar Int
committedGeneration <- IO (TMVar Int)
forall a. IO (TMVar a)
newEmptyTMVarIO
    SID
inbox <- ClientState -> IO SID
nextInbox ClientState
client
    let subscriptionConfig :: SubscribeConfig
subscriptionConfig = Maybe NominalDiffTime -> Maybe SID -> SubscribeConfig
SubscribeConfig Maybe NominalDiffTime
forall a. Maybe a
Nothing Maybe SID
forall a. Maybe a
Nothing
        deliver :: Maybe Msg -> IO ()
deliver Maybe Msg
Nothing = STM () -> IO ()
forall a. STM a -> IO a
atomically (STM Bool -> STM ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (TMVar (Either NatsError Message)
-> Either NatsError Message -> STM Bool
forall a. TMVar a -> a -> STM Bool
tryPutTMVar TMVar (Either NatsError Message)
response (NatsError -> Either NatsError Message
forall a b. a -> Either a b
Left NatsError
NatsRequestTimedOut)))
        deliver (Just Msg
msg) =
          STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ())
-> (Either NatsError Message -> STM ())
-> Either NatsError Message
-> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. STM Bool -> STM ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (STM Bool -> STM ())
-> (Either NatsError Message -> STM Bool)
-> Either NatsError Message
-> STM ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. TMVar (Either NatsError Message)
-> Either NatsError Message -> STM Bool
forall a. TMVar a -> a -> STM Bool
tryPutTMVar TMVar (Either NatsError Message)
response (Either NatsError Message -> IO ())
-> Either NatsError Message -> IO ()
forall a b. (a -> b) -> a -> b
$
            let message :: Message
message = Msg -> Message
toMessage Msg
msg
            in if Message -> Bool
isNoResponders Message
message
                 then NatsError -> Either NatsError Message
forall a b. a -> Either a b
Left NatsError
NatsNoResponders
                 else Message -> Either NatsError Message
forall a b. b -> Either a b
Right Message
message
        rejectSlowConsumer :: STM ()
rejectSlowConsumer =
          STM Bool -> STM ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (TMVar (Either NatsError Message)
-> Either NatsError Message -> STM Bool
forall a. TMVar a -> a -> STM Bool
tryPutTMVar TMVar (Either NatsError Message)
response (NatsError -> Either NatsError Message
forall a b. a -> Either a b
Left NatsError
NatsSlowConsumer))
    SID
sid <- ClientState -> IO SID
nextSid ClientState
client
    let subscription :: Subscription
subscription = SID -> Subscription
Subscription SID
sid
        subscriptionMessage :: Sub
subscriptionMessage =
          Sub.Sub
            { subject :: SID
Sub.subject = SID
inbox
            , queueGroup :: Maybe SID
Sub.queueGroup = Maybe SID
forall a. Maybe a
Nothing
            , sid :: SID
Sub.sid = SID
sid
            }
        unsubscribeMessage :: Unsub
unsubscribeMessage =
          Unsub.Unsub
            { sid :: SID
Unsub.sid = SID
sid
            , maxMsg :: Maybe Int
Unsub.maxMsg = Int -> Maybe Int
forall a. a -> Maybe a
Just Int
1
            }
        publishMessage :: Pub
publishMessage =
          Pub.Pub
            { subject :: SID
Pub.subject = SID
requestSubject
            , payload :: Maybe SID
Pub.payload =
                if SID -> Bool
BS.null SID
requestPayload then Maybe SID
forall a. Maybe a
Nothing else SID -> Maybe SID
forall a. a -> Maybe a
Just SID
requestPayload
            , replyTo :: Maybe SID
Pub.replyTo = SID -> Maybe SID
forall a. a -> Maybe a
Just SID
inbox
            , headers :: Maybe Headers
Pub.headers = RequestConfig -> Maybe Headers
requestHeaders RequestConfig
cfg
            }
        commands :: QueueItem
commands = QueueItem -> QueueItem
QueueConnectionScoped (QueueItem -> QueueItem)
-> ([QueueItem] -> QueueItem) -> [QueueItem] -> QueueItem
forall b c a. (b -> c) -> (a -> b) -> a -> c
. [QueueItem] -> QueueItem
QueueBatch ([QueueItem] -> QueueItem) -> [QueueItem] -> QueueItem
forall a b. (a -> b) -> a -> b
$
          [ Sub -> QueueItem
forall m. Transformer m => m -> QueueItem
QueueItem Sub
subscriptionMessage
          , Unsub -> QueueItem
forall m. Transformer m => m -> QueueItem
QueueItem Unsub
unsubscribeMessage
          , Pub -> QueueItem
forall m. Transformer m => m -> QueueItem
QueueItem Pub
publishMessage
          ]
        meta :: SubscriptionMeta
meta = SID -> Maybe SID -> SubscriptionKind -> SubscriptionMeta
SubscriptionMeta SID
inbox Maybe SID
forall a. Maybe a
Nothing SubscriptionKind
RequestReplySubscription
    case Sub -> Either SID ()
forall a. Validator a => a -> Either SID ()
validate Sub
subscriptionMessage of
      Left SID
reason -> Either NatsError Message -> IO (Either NatsError Message)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (NatsError -> Either NatsError Message
forall a b. a -> Either a b
Left (SID -> NatsError
NatsValidationError SID
reason))
      Right () -> do
        Either NatsError ()
publishValidation <- ClientState -> Pub -> IO (Either NatsError ())
validatePublish ClientState
client Pub
publishMessage
        case Either NatsError ()
publishValidation of
          Left NatsError
err -> Either NatsError Message -> IO (Either NatsError Message)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (NatsError -> Either NatsError Message
forall a b. a -> Either a b
Left NatsError
err)
          Right () -> do
            SubscriptionStore
-> SID
-> SubscriptionMeta
-> SubscribeConfig
-> STM ()
-> STM ()
-> (Maybe Msg -> IO ())
-> IO ()
-> IO ()
registerWithDispatchHooks
              SubscriptionStore
store
              SID
sid
              SubscriptionMeta
meta
              SubscribeConfig
subscriptionConfig
              (STM Bool -> STM ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (TMVar () -> () -> STM Bool
forall a. TMVar a -> a -> STM Bool
tryPutTMVar TMVar ()
accepted ()))
              STM ()
rejectSlowConsumer
              Maybe Msg -> IO ()
deliver
              (() -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ())
            let cleanupCommitted :: IO ()
cleanupCommitted = do
                  Maybe Int
committed <- STM (Maybe Int) -> IO (Maybe Int)
forall a. STM a -> IO a
atomically (TMVar Int -> STM (Maybe Int)
forall a. TMVar a -> STM (Maybe a)
tryReadTMVar TMVar Int
committedGeneration)
                  case Maybe Int
committed of
                    Maybe Int
Nothing -> SubscriptionStore -> SID -> IO ()
unregister SubscriptionStore
store SID
sid
                    Just Int
generation ->
                      ClientState -> SubscriptionStore -> Int -> Subscription -> IO ()
cleanupRequestSubscription ClientState
client SubscriptionStore
store Int
generation Subscription
subscription
            let publishSize :: Int
publishSize = Pub -> Int
Pub.messageSize Pub
publishMessage
            PublishEnqueueResult
queued <- IO PublishEnqueueResult -> IO PublishEnqueueResult
forall a. IO a -> IO a
restore
              ( ClientState
-> TMVar Int -> Int -> QueueItem -> IO PublishEnqueueResult
enqueuePublishOnConnectedGenerationTracked
                  ClientState
client
                  TMVar Int
committedGeneration
                  Int
publishSize
                  QueueItem
commands
              )
              IO PublishEnqueueResult -> IO () -> IO PublishEnqueueResult
forall a b. IO a -> IO b -> IO a
`onException` IO ()
cleanupCommitted
            case PublishEnqueueResult
queued of
              PublishConnectionClosed ClientExitReason
reason -> do
                SubscriptionStore -> SID -> IO ()
unregister SubscriptionStore
store SID
sid
                Either NatsError Message -> IO (Either NatsError Message)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (NatsError -> Either NatsError Message
forall a b. a -> Either a b
Left (ClientExitReason -> NatsError
NatsConnectionClosed ClientExitReason
reason))
              PublishTooLarge Int
maximumSize -> do
                SubscriptionStore -> SID -> IO ()
unregister SubscriptionStore
store SID
sid
                ClientState -> Int -> Int -> IO (Either NatsError Message)
forall a. ClientState -> Int -> Int -> IO (Either NatsError a)
rejectPayloadTooLarge ClientState
client Int
publishSize Int
maximumSize
              PublishEnqueued Int
generation -> do
                let cleanup :: IO ()
cleanup =
                      ClientState -> SubscriptionStore -> Int -> Subscription -> IO ()
cleanupRequestSubscription ClientState
client SubscriptionStore
store Int
generation Subscription
subscription
                Either NatsError Message
result <- IO (Either NatsError Message) -> IO (Either NatsError Message)
forall a. IO a -> IO a
restore (ClientState
-> Int
-> TMVar ()
-> TMVar (Either NatsError Message)
-> IO (Either NatsError Message)
awaitRequest ClientState
client Int
generation TMVar ()
accepted TMVar (Either NatsError Message)
response)
                  IO (Either NatsError Message)
-> IO () -> IO (Either NatsError Message)
forall a b. IO a -> IO b -> IO a
`onException` IO ()
cleanup
                IO ()
cleanup
                Either NatsError Message -> IO (Either NatsError Message)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Either NatsError Message
result

awaitRequest
  :: ClientState
  -> Int
  -> TMVar ()
  -> TMVar (Either NatsError Message)
  -> IO (Either NatsError Message)
awaitRequest :: ClientState
-> Int
-> TMVar ()
-> TMVar (Either NatsError Message)
-> IO (Either NatsError Message)
awaitRequest ClientState
client Int
generation TMVar ()
accepted TMVar (Either NatsError Message)
response =
  STM (Either NatsError Message) -> IO (Either NatsError Message)
forall a. STM a -> IO a
atomically (STM (Either NatsError Message) -> IO (Either NatsError Message))
-> STM (Either NatsError Message) -> IO (Either NatsError Message)
forall a b. (a -> b) -> a -> b
$
    TMVar (Either NatsError Message) -> STM (Either NatsError Message)
forall a. TMVar a -> STM a
readTMVar TMVar (Either NatsError Message)
response
      STM (Either NatsError Message)
-> STM (Either NatsError Message) -> STM (Either NatsError Message)
forall a. STM a -> STM a -> STM a
`orElse`
        do
          Maybe ()
acceptedReply <- TMVar () -> STM (Maybe ())
forall a. TMVar a -> STM (Maybe a)
tryReadTMVar TMVar ()
accepted
          case Maybe ()
acceptedReply of
            Just () -> STM (Either NatsError Message)
forall a. STM a
retry
            Maybe ()
Nothing ->
              NatsError -> Either NatsError Message
forall a b. a -> Either a b
Left (NatsError -> Either NatsError Message)
-> (ClientExitReason -> NatsError)
-> ClientExitReason
-> Either NatsError Message
forall b c a. (b -> c) -> (a -> b) -> a -> c
. ClientExitReason -> NatsError
NatsConnectionClosed
                (ClientExitReason -> Either NatsError Message)
-> STM ClientExitReason -> STM (Either NatsError Message)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> ClientState -> Int -> STM ClientExitReason
waitForConnectionGenerationLoss ClientState
client Int
generation

durationMicros :: NominalDiffTime -> Int
durationMicros :: NominalDiffTime -> Int
durationMicros NominalDiffTime
duration =
  Integer -> Int
forall a. Num a => Integer -> a
fromInteger (Integer -> Integer -> Integer
forall a. Ord a => a -> a -> a
min (Int -> Integer
forall a. Integral a => a -> Integer
toInteger (Int
forall a. Bounded a => a
maxBound :: Int)) Integer
micros)
  where
    micros :: Integer
micros = Integer -> Integer -> Integer
forall a. Ord a => a -> a -> a
max Integer
0 (Rational -> Integer
forall b. Integral b => Rational -> b
forall a b. (RealFrac a, Integral b) => a -> b
floor (NominalDiffTime -> Rational
forall a. Real a => a -> Rational
toRational NominalDiffTime
duration Rational -> Rational -> Rational
forall a. Num a => a -> a -> a
* Rational
1000000))

isNoResponders :: Message -> Bool
isNoResponders :: Message -> Bool
isNoResponders Message
message =
  case Message -> Maybe Headers
headers Message
message of
    Maybe Headers
Nothing -> Bool
False
    Just Headers
messageHeaders ->
      (UserPassData -> Bool) -> Headers -> Bool
forall (t :: * -> *) a. Foldable t => (a -> Bool) -> t a -> Bool
any UserPassData -> Bool
forall {a}. (Eq a, IsString a) => (SID, a) -> Bool
isNoRespondersHeader Headers
messageHeaders
  where
    isNoRespondersHeader :: (SID, a) -> Bool
isNoRespondersHeader (SID
name, a
value) =
      (Char -> Char) -> SID -> SID
BC.map Char -> Char
toAsciiLower SID
name SID -> SID -> Bool
forall a. Eq a => a -> a -> Bool
== SID
"status" Bool -> Bool -> Bool
&& a
value a -> a -> Bool
forall a. Eq a => a -> a -> Bool
== a
"503"
    toAsciiLower :: Char -> Char
toAsciiLower Char
byte
      | Char -> Bool
isAsciiUpper Char
byte = Int -> Char
forall a. Enum a => Int -> a
toEnum (Char -> Int
forall a. Enum a => a -> Int
fromEnum Char
byte Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
32)
      | Bool
otherwise = Char
byte

unsubscribeClient
  :: ClientState
  -> SubscriptionStore
  -> Subscription
  -> IO (Either NatsError ())
unsubscribeClient :: ClientState
-> SubscriptionStore -> Subscription -> IO (Either NatsError ())
unsubscribeClient ClientState
client SubscriptionStore
store (Subscription SID
sid) =
  ((forall a. IO a -> IO a) -> IO (Either NatsError ()))
-> IO (Either NatsError ())
forall b. ((forall a. IO a -> IO a) -> IO b) -> IO b
mask (((forall a. IO a -> IO a) -> IO (Either NatsError ()))
 -> IO (Either NatsError ()))
-> ((forall a. IO a -> IO a) -> IO (Either NatsError ()))
-> IO (Either NatsError ())
forall a b. (a -> b) -> a -> b
$ \forall a. IO a -> IO a
restore -> do
    ClientState -> AppM () -> IO ()
forall a. ClientState -> AppM a -> IO a
runClient ClientState
client (AppM () -> IO ()) -> AppM () -> IO ()
forall a b. (a -> b) -> a -> b
$
      LogLevel -> [Char] -> AppM ()
forall (m :: * -> *). MonadLogger m => LogLevel -> [Char] -> m ()
logMessage LogLevel
Debug ([Char]
"unsubscribing SID: " [Char] -> ShowS
forall a. [a] -> [a] -> [a]
++ SID -> [Char]
forall a. Show a => a -> [Char]
show SID
sid)
    let command :: QueueItem
command = QueueItem -> QueueItem
QueueConnectionScoped (QueueItem -> QueueItem)
-> (Unsub -> QueueItem) -> Unsub -> QueueItem
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Unsub -> QueueItem
forall m. Transformer m => m -> QueueItem
QueueItem (Unsub -> QueueItem) -> Unsub -> QueueItem
forall a b. (a -> b) -> a -> b
$
          Unsub.Unsub
            { sid :: SID
Unsub.sid = SID
sid
            , maxMsg :: Maybe Int
Unsub.maxMsg = Maybe Int
forall a. Maybe a
Nothing
            }
        unregisterAndTry :: IO (Either a UnsubscribeQueueResult)
unregisterAndTry = do
          SubscriptionStore -> SID -> IO ()
unregister SubscriptionStore
store SID
sid
          ClientState -> QueueItem -> IO TryEnqueueResult
tryEnqueue ClientState
client QueueItem
command IO TryEnqueueResult
-> (TryEnqueueResult -> IO (Either a UnsubscribeQueueResult))
-> IO (Either a UnsubscribeQueueResult)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
            TryEnqueueResult
TryEnqueued    -> Either a UnsubscribeQueueResult
-> IO (Either a UnsubscribeQueueResult)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (UnsubscribeQueueResult -> Either a UnsubscribeQueueResult
forall a b. b -> Either a b
Right UnsubscribeQueueResult
UnsubscribeQueued)
            TryEnqueueResult
TryQueueClosed -> Either a UnsubscribeQueueResult
-> IO (Either a UnsubscribeQueueResult)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (UnsubscribeQueueResult -> Either a UnsubscribeQueueResult
forall a b. b -> Either a b
Right UnsubscribeQueueResult
UnsubscribeReconnect)
            TryEnqueueResult
TryQueueFull   -> Either a UnsubscribeQueueResult
-> IO (Either a UnsubscribeQueueResult)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (UnsubscribeQueueResult -> Either a UnsubscribeQueueResult
forall a b. b -> Either a b
Right UnsubscribeQueueResult
UnsubscribeReset)
    Either NatsError UnsubscribeQueueResult
queueResult <- ClientState
-> IO (Either NatsError UnsubscribeQueueResult)
-> IO (Either NatsError UnsubscribeQueueResult)
forall a. ClientState -> IO a -> IO a
withSubscriptionGate ClientState
client (IO (Either NatsError UnsubscribeQueueResult)
 -> IO (Either NatsError UnsubscribeQueueResult))
-> IO (Either NatsError UnsubscribeQueueResult)
-> IO (Either NatsError UnsubscribeQueueResult)
forall a b. (a -> b) -> a -> b
$ do
      ConnectionState
status <- ClientState -> IO ConnectionState
readStatus ClientState
client
      case ConnectionState
status of
        ConnectionState
ConnectionConnected -> do
          SubscriptionStore -> SID -> IO ()
unregister SubscriptionStore
store SID
sid
          Either [Char] ()
enqueueResult <-
            IO (Either [Char] ()) -> IO (Either [Char] ())
forall a. IO a -> IO a
restore (ClientState -> QueueItem -> IO (Either [Char] ())
enqueue ClientState
client QueueItem
command)
              IO (Either [Char] ()) -> IO Int -> IO (Either [Char] ())
forall a b. IO a -> IO b -> IO a
`onException` ConnectionAPI -> ClientState -> IO Int
interruptConnection ConnectionAPI
connectionApi ClientState
client
          case Either [Char] ()
enqueueResult of
            Right () -> Either NatsError UnsubscribeQueueResult
-> IO (Either NatsError UnsubscribeQueueResult)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (UnsubscribeQueueResult -> Either NatsError UnsubscribeQueueResult
forall a b. b -> Either a b
Right UnsubscribeQueueResult
UnsubscribeQueued)
            Left [Char]
_ -> do
              ConnectionState
nextStatus <- ClientState -> IO ConnectionState
readStatus ClientState
client
              case ConnectionState
nextStatus of
                ConnectionClosing ClientExitReason
reason ->
                  Either NatsError UnsubscribeQueueResult
-> IO (Either NatsError UnsubscribeQueueResult)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (NatsError -> Either NatsError UnsubscribeQueueResult
forall a b. a -> Either a b
Left (ClientExitReason -> NatsError
NatsConnectionClosed ClientExitReason
reason))
                ConnectionClosed ClientExitReason
reason ->
                  Either NatsError UnsubscribeQueueResult
-> IO (Either NatsError UnsubscribeQueueResult)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (NatsError -> Either NatsError UnsubscribeQueueResult
forall a b. a -> Either a b
Left (ClientExitReason -> NatsError
NatsConnectionClosed ClientExitReason
reason))
                ConnectionState
_ -> Either NatsError UnsubscribeQueueResult
-> IO (Either NatsError UnsubscribeQueueResult)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (UnsubscribeQueueResult -> Either NatsError UnsubscribeQueueResult
forall a b. b -> Either a b
Right UnsubscribeQueueResult
UnsubscribeReconnect)
        ConnectionState
ConnectionConnecting -> IO (Either NatsError UnsubscribeQueueResult)
forall {a}. IO (Either a UnsubscribeQueueResult)
unregisterAndTry
        ConnectionState
ConnectionReconnecting -> IO (Either NatsError UnsubscribeQueueResult)
forall {a}. IO (Either a UnsubscribeQueueResult)
unregisterAndTry
        ConnectionClosing ClientExitReason
reason -> do
          SubscriptionStore -> SID -> IO ()
unregister SubscriptionStore
store SID
sid
          Either NatsError UnsubscribeQueueResult
-> IO (Either NatsError UnsubscribeQueueResult)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (NatsError -> Either NatsError UnsubscribeQueueResult
forall a b. a -> Either a b
Left (ClientExitReason -> NatsError
NatsConnectionClosed ClientExitReason
reason))
        ConnectionClosed ClientExitReason
reason -> do
          SubscriptionStore -> SID -> IO ()
unregister SubscriptionStore
store SID
sid
          Either NatsError UnsubscribeQueueResult
-> IO (Either NatsError UnsubscribeQueueResult)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (NatsError -> Either NatsError UnsubscribeQueueResult
forall a b. a -> Either a b
Left (ClientExitReason -> NatsError
NatsConnectionClosed ClientExitReason
reason))
    case Either NatsError UnsubscribeQueueResult
queueResult of
      Left NatsError
err                    -> Either NatsError () -> IO (Either NatsError ())
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (NatsError -> Either NatsError ()
forall a b. a -> Either a b
Left NatsError
err)
      Right UnsubscribeQueueResult
UnsubscribeQueued    -> Either NatsError () -> IO (Either NatsError ())
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (() -> Either NatsError ()
forall a b. b -> Either a b
Right ())
      Right UnsubscribeQueueResult
UnsubscribeReconnect -> Either NatsError () -> IO (Either NatsError ())
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (() -> Either NatsError ()
forall a b. b -> Either a b
Right ())
      Right UnsubscribeQueueResult
UnsubscribeReset     -> do
        ConnectionAPI -> ClientState -> IO Int
interruptConnection ConnectionAPI
connectionApi ClientState
client
        Either NatsError () -> IO (Either NatsError ())
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (() -> Either NatsError ()
forall a b. b -> Either a b
Right ())

-- | A drain keeps callbacks registered until NATS confirms every preceding
-- @UNSUB@. The PONG is an ordering fence: all messages the server could have
-- delivered before the unsubscribe are now queued locally, so waiting on their
-- subscriptions cannot race a reply against connection close.
drainSubscriptionsClient
  :: ConnectionAPI
  -> ClientState
  -> SubscriptionStore
  -> [Subscription]
  -> NominalDiffTime
  -> IO (Either NatsError ())
drainSubscriptionsClient :: ConnectionAPI
-> ClientState
-> SubscriptionStore
-> [Subscription]
-> NominalDiffTime
-> IO (Either NatsError ())
drainSubscriptionsClient ConnectionAPI
_ ClientState
_ SubscriptionStore
_ [] NominalDiffTime
_ =
  Either NatsError () -> IO (Either NatsError ())
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (() -> Either NatsError ()
forall a b. b -> Either a b
Right ())
drainSubscriptionsClient ConnectionAPI
connectionApi' ClientState
client SubscriptionStore
store [Subscription]
subscriptions NominalDiffTime
timeoutSeconds = do
  Either NatsError ()
fenced <- IO (Either NatsError ())
awaitFence
  case Either NatsError ()
fenced of
    Left NatsError
err ->
      Either NatsError () -> IO (Either NatsError ())
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (NatsError -> Either NatsError ()
forall a b. a -> Either a b
Left NatsError
err)
    Right () -> do
      STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$
        (Subscription -> STM ()) -> [Subscription] -> STM ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
(a -> m b) -> t a -> m ()
mapM_ (SubscriptionStore -> SID -> STM ()
awaitSubscriptionDrain SubscriptionStore
store (SID -> STM ()) -> (Subscription -> SID) -> Subscription -> STM ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Subscription -> SID
subscriptionSid) [Subscription]
subscriptions
      (Subscription -> IO ()) -> [Subscription] -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
(a -> m b) -> t a -> m ()
mapM_ (SubscriptionStore -> SID -> IO ()
unregister SubscriptionStore
store (SID -> IO ()) -> (Subscription -> SID) -> Subscription -> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Subscription -> SID
subscriptionSid) [Subscription]
subscriptions
      Either NatsError () -> IO (Either NatsError ())
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (() -> Either NatsError ()
forall a b. b -> Either a b
Right ())
  where
    commands :: QueueItem
commands =
      QueueItem -> QueueItem
QueueConnectionScoped (QueueItem -> QueueItem)
-> ([QueueItem] -> QueueItem) -> [QueueItem] -> QueueItem
forall b c a. (b -> c) -> (a -> b) -> a -> c
. [QueueItem] -> QueueItem
QueueBatch ([QueueItem] -> QueueItem) -> [QueueItem] -> QueueItem
forall a b. (a -> b) -> a -> b
$
        (Subscription -> QueueItem) -> [Subscription] -> [QueueItem]
forall a b. (a -> b) -> [a] -> [b]
map Subscription -> QueueItem
unsubscribeCommand [Subscription]
subscriptions

    unsubscribeCommand :: Subscription -> QueueItem
unsubscribeCommand (Subscription SID
sid) =
      Unsub -> QueueItem
forall m. Transformer m => m -> QueueItem
QueueItem Unsub.Unsub { sid :: SID
Unsub.sid = SID
sid, maxMsg :: Maybe Int
Unsub.maxMsg = Maybe Int
forall a. Maybe a
Nothing }

    awaitFence :: IO (Either NatsError ())
awaitFence = do
      Maybe (Either NatsError ())
result <- ClientState
-> IO (Maybe (Either NatsError ()))
-> IO (Maybe (Either NatsError ()))
forall a. ClientState -> IO a -> IO a
withSubscriptionGate ClientState
client (IO (Maybe (Either NatsError ()))
 -> IO (Maybe (Either NatsError ())))
-> IO (Maybe (Either NatsError ()))
-> IO (Maybe (Either NatsError ()))
forall a b. (a -> b) -> a -> b
$ do
        ConnectionState
status <- ClientState -> IO ConnectionState
readStatus ClientState
client
        case ConnectionState
status of
          ConnectionState
ConnectionConnected ->
            Either NatsError () -> Maybe (Either NatsError ())
forall a. a -> Maybe a
Just (Either NatsError () -> Maybe (Either NatsError ()))
-> IO (Either NatsError ()) -> IO (Maybe (Either NatsError ()))
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> IO (Either NatsError ())
fenceConnected
          ConnectionState
ConnectionConnecting ->
            Maybe (Either NatsError ()) -> IO (Maybe (Either NatsError ()))
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Maybe (Either NatsError ())
forall a. Maybe a
Nothing
          ConnectionState
ConnectionReconnecting ->
            Maybe (Either NatsError ()) -> IO (Maybe (Either NatsError ()))
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Maybe (Either NatsError ())
forall a. Maybe a
Nothing
          ConnectionClosing ClientExitReason
reason ->
            Maybe (Either NatsError ()) -> IO (Maybe (Either NatsError ()))
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Either NatsError () -> Maybe (Either NatsError ())
forall a. a -> Maybe a
Just (NatsError -> Either NatsError ()
forall a b. a -> Either a b
Left (ClientExitReason -> NatsError
NatsConnectionClosed ClientExitReason
reason)))
          ConnectionClosed ClientExitReason
reason ->
            Maybe (Either NatsError ()) -> IO (Maybe (Either NatsError ()))
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Either NatsError () -> Maybe (Either NatsError ())
forall a. a -> Maybe a
Just (NatsError -> Either NatsError ()
forall a b. a -> Either a b
Left (ClientExitReason -> NatsError
NatsConnectionClosed ClientExitReason
reason)))
      case Maybe (Either NatsError ())
result of
        Just Either NatsError ()
value ->
          Either NatsError () -> IO (Either NatsError ())
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Either NatsError ()
value
        Maybe (Either NatsError ())
Nothing -> do
          Either NatsError ()
connected <- ClientState -> IO (Either NatsError ())
runningResult ClientState
client
          case Either NatsError ()
connected of
            Left NatsError
err -> Either NatsError () -> IO (Either NatsError ())
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (NatsError -> Either NatsError ()
forall a b. a -> Either a b
Left NatsError
err)
            Right () -> IO (Either NatsError ())
awaitFence

    fenceConnected :: IO (Either NatsError ())
fenceConnected = do
      Either [Char] ()
enqueueResult <- ClientState -> QueueItem -> IO (Either [Char] ())
enqueue ClientState
client QueueItem
commands
      case Either [Char] ()
enqueueResult of
        Right () -> do
          Either NatsError ()
flushed <- ConnectionAPI
-> ClientState -> NominalDiffTime -> IO (Either NatsError ())
flushClient ConnectionAPI
connectionApi' ClientState
client NominalDiffTime
timeoutSeconds
          case Either NatsError ()
flushed of
            Left NatsError
err ->
              Either NatsError () -> IO (Either NatsError ())
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (NatsError -> Either NatsError ()
forall a b. a -> Either a b
Left NatsError
err)
            Right () -> do
              (Subscription -> IO ()) -> [Subscription] -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
(a -> m b) -> t a -> m ()
mapM_ (SubscriptionStore -> SID -> IO ()
retire SubscriptionStore
store (SID -> IO ()) -> (Subscription -> SID) -> Subscription -> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Subscription -> SID
subscriptionSid) [Subscription]
subscriptions
              Either NatsError () -> IO (Either NatsError ())
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (() -> Either NatsError ()
forall a b. b -> Either a b
Right ())
        Left [Char]
_ -> do
          ConnectionState
status <- ClientState -> IO ConnectionState
readStatus ClientState
client
          Either NatsError () -> IO (Either NatsError ())
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (NatsError -> Either NatsError ()
forall a b. a -> Either a b
Left (ConnectionState -> NatsError
closedError ConnectionState
status))

cleanupResumableSubscription
  :: ClientState
  -> SubscriptionStore
  -> TMVar Int
  -> TMVar Int
  -> Subscription
  -> IO ()
cleanupResumableSubscription :: ClientState
-> SubscriptionStore
-> TMVar Int
-> TMVar Int
-> Subscription
-> IO ()
cleanupResumableSubscription ClientState
client SubscriptionStore
store TMVar Int
registeredGeneration TMVar Int
committedGeneration (Subscription SID
sid) =
  IO () -> IO ()
forall a. IO a -> IO a
mask_ (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
    Bool
resetRequired <- ClientState -> IO Bool -> IO Bool
forall a. ClientState -> IO a -> IO a
withSubscriptionGate ClientState
client (IO Bool -> IO Bool) -> IO Bool -> IO Bool
forall a b. (a -> b) -> a -> b
$ do
      SubscriptionStore -> SID -> IO ()
unregister SubscriptionStore
store SID
sid
      ConnectionState
status <- ClientState -> IO ConnectionState
readStatus ClientState
client
      case ConnectionState
status of
        ConnectionState
ConnectionConnected -> do
          Int
generation <- ClientState -> IO Int
readConnectionGeneration ClientState
client
          Maybe Int
registered <- STM (Maybe Int) -> IO (Maybe Int)
forall a. STM a -> IO a
atomically (TMVar Int -> STM (Maybe Int)
forall a. TMVar a -> STM (Maybe a)
tryReadTMVar TMVar Int
registeredGeneration)
          Maybe Int
committed <- STM (Maybe Int) -> IO (Maybe Int)
forall a. STM a -> IO a
atomically (TMVar Int -> STM (Maybe Int)
forall a. TMVar a -> STM (Maybe a)
tryReadTMVar TMVar Int
committedGeneration)
          let serverMayKnow :: Bool
serverMayKnow =
                Maybe Int -> Bool
forall a. Maybe a -> Bool
isJust Maybe Int
committed Bool -> Bool -> Bool
|| Bool -> (Int -> Bool) -> Maybe Int -> Bool
forall b a. b -> (a -> b) -> Maybe a -> b
maybe Bool
False (Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
< Int
generation) Maybe Int
registered
          if Bool
serverMayKnow
            then do
              TryEnqueueResult
result <- ClientState -> Int -> QueueItem -> IO TryEnqueueResult
tryEnqueueOnGeneration ClientState
client Int
generation QueueItem
unsubscribeCommand
              Bool -> IO Bool
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (TryEnqueueResult
result TryEnqueueResult -> TryEnqueueResult -> Bool
forall a. Eq a => a -> a -> Bool
== TryEnqueueResult
TryQueueFull)
            else Bool -> IO Bool
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Bool
False
        ConnectionState
_ -> Bool -> IO Bool
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Bool
False
    Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
resetRequired (IO Int -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (ConnectionAPI -> ClientState -> IO Int
interruptConnection ConnectionAPI
connectionApi ClientState
client))
  where
    unsubscribeCommand :: QueueItem
unsubscribeCommand = QueueItem -> QueueItem
QueueConnectionScoped (QueueItem -> QueueItem)
-> (Unsub -> QueueItem) -> Unsub -> QueueItem
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Unsub -> QueueItem
forall m. Transformer m => m -> QueueItem
QueueItem (Unsub -> QueueItem) -> Unsub -> QueueItem
forall a b. (a -> b) -> a -> b
$
      Unsub.Unsub { sid :: SID
Unsub.sid = SID
sid, maxMsg :: Maybe Int
Unsub.maxMsg = Maybe Int
forall a. Maybe a
Nothing }

cleanupRequestSubscription :: ClientState -> SubscriptionStore -> Int -> Subscription -> IO ()
cleanupRequestSubscription :: ClientState -> SubscriptionStore -> Int -> Subscription -> IO ()
cleanupRequestSubscription ClientState
client SubscriptionStore
store Int
generation (Subscription SID
sid) = IO () -> IO ()
forall a. IO a -> IO a
mask_ (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
  SubscriptionStore -> SID -> IO ()
unregister SubscriptionStore
store SID
sid
  TryEnqueueResult
result <- ClientState -> Int -> QueueItem -> IO TryEnqueueResult
tryEnqueueOnGeneration ClientState
client Int
generation (QueueItem -> IO TryEnqueueResult)
-> (Unsub -> QueueItem) -> Unsub -> IO TryEnqueueResult
forall b c a. (b -> c) -> (a -> b) -> a -> c
. QueueItem -> QueueItem
QueueConnectionScoped (QueueItem -> QueueItem)
-> (Unsub -> QueueItem) -> Unsub -> QueueItem
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Unsub -> QueueItem
forall m. Transformer m => m -> QueueItem
QueueItem (Unsub -> IO TryEnqueueResult) -> Unsub -> IO TryEnqueueResult
forall a b. (a -> b) -> a -> b
$
    Unsub.Unsub { sid :: SID
Unsub.sid = SID
sid, maxMsg :: Maybe Int
Unsub.maxMsg = Maybe Int
forall a. Maybe a
Nothing }
  case TryEnqueueResult
result of
    TryEnqueueResult
TryEnqueued    -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
    TryEnqueueResult
TryQueueClosed -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
    TryEnqueueResult
TryQueueFull   -> IO Int -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (ConnectionAPI -> ClientState -> IO Int
interruptConnection ConnectionAPI
connectionApi ClientState
client)

flushClient :: ConnectionAPI -> ClientState -> NominalDiffTime -> IO (Either NatsError ())
flushClient :: ConnectionAPI
-> ClientState -> NominalDiffTime -> IO (Either NatsError ())
flushClient ConnectionAPI
connectionApi' ClientState
client NominalDiffTime
timeoutSeconds =
  ((forall a. IO a -> IO a) -> IO (Either NatsError ()))
-> IO (Either NatsError ())
forall b. ((forall a. IO a -> IO a) -> IO b) -> IO b
mask (((forall a. IO a -> IO a) -> IO (Either NatsError ()))
 -> IO (Either NatsError ()))
-> ((forall a. IO a -> IO a) -> IO (Either NatsError ()))
-> IO (Either NatsError ())
forall a b. (a -> b) -> a -> b
$ \forall a. IO a -> IO a
restore -> do
    let reset :: IO Int
reset = ConnectionAPI -> ClientState -> IO Int
interruptConnection ConnectionAPI
connectionApi' ClientState
client
    Maybe (Either NatsError ())
timedResult <-
      IO (Maybe (Either NatsError ()))
-> IO (Maybe (Either NatsError ()))
forall a. IO a -> IO a
restore (Int -> IO (Either NatsError ()) -> IO (Maybe (Either NatsError ()))
forall a. Int -> IO a -> IO (Maybe a)
timeout (NominalDiffTime -> Int
durationMicros NominalDiffTime
timeoutSeconds) (ClientState -> IO (Either NatsError ())
awaitPong ClientState
client))
        IO (Maybe (Either NatsError ()))
-> IO Int -> IO (Maybe (Either NatsError ()))
forall a b. IO a -> IO b -> IO a
`onException` IO Int
reset
    case Maybe (Either NatsError ())
timedResult of
      Maybe (Either NatsError ())
Nothing     -> IO Int
reset IO Int -> IO (Either NatsError ()) -> IO (Either NatsError ())
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> Either NatsError () -> IO (Either NatsError ())
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (NatsError -> Either NatsError ()
forall a b. a -> Either a b
Left NatsError
NatsRequestTimedOut)
      Just Either NatsError ()
result -> Either NatsError () -> IO (Either NatsError ())
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Either NatsError ()
result
  where
    awaitPong :: ClientState -> IO (Either NatsError ())
awaitPong ClientState
client' = do
      TMVar PingResult
waiter <- IO (TMVar PingResult)
forall a. IO (TMVar a)
newEmptyTMVarIO
      Either ClientExitReason Int
registered <- ClientState
-> TMVar PingResult
-> QueueItem
-> IO (Either ClientExitReason Int)
registerPingWaiterAndEnqueue
        ClientState
client'
        TMVar PingResult
waiter
        (QueueItem -> QueueItem
QueueConnectionScoped (Ping -> QueueItem
forall m. Transformer m => m -> QueueItem
QueueItem Ping
Ping))
      case Either ClientExitReason Int
registered of
        Left ClientExitReason
reason -> Either NatsError () -> IO (Either NatsError ())
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (NatsError -> Either NatsError ()
forall a b. a -> Either a b
Left (ClientExitReason -> NatsError
NatsConnectionClosed ClientExitReason
reason))
        Right Int
_ -> do
          Maybe PingResult
result <- STM (Maybe PingResult) -> IO (Maybe PingResult)
forall a. STM a -> IO a
atomically (STM (Maybe PingResult) -> IO (Maybe PingResult))
-> STM (Maybe PingResult) -> IO (Maybe PingResult)
forall a b. (a -> b) -> a -> b
$
            (PingResult -> Maybe PingResult
forall a. a -> Maybe a
Just (PingResult -> Maybe PingResult)
-> STM PingResult -> STM (Maybe PingResult)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> TMVar PingResult -> STM PingResult
forall a. TMVar a -> STM a
readTMVar TMVar PingResult
waiter)
              STM (Maybe PingResult)
-> STM (Maybe PingResult) -> STM (Maybe PingResult)
forall a. STM a -> STM a -> STM a
`orElse` (Maybe PingResult
forall a. Maybe a
Nothing Maybe PingResult -> STM () -> STM (Maybe PingResult)
forall a b. a -> STM b -> STM a
forall (f :: * -> *) a b. Functor f => a -> f b -> f a
<$ ClientState -> STM ()
waitForNotRunning ClientState
client')
          case Maybe PingResult
result of
            Just PingResult
PingReceived -> Either NatsError () -> IO (Either NatsError ())
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (() -> Either NatsError ()
forall a b. b -> Either a b
Right ())
            Just PingResult
PingConnectionLost ->
              NatsError -> Either NatsError ()
forall a b. a -> Either a b
Left (NatsError -> Either NatsError ())
-> (ConnectionState -> NatsError)
-> ConnectionState
-> Either NatsError ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. ConnectionState -> NatsError
closedError (ConnectionState -> Either NatsError ())
-> IO ConnectionState -> IO (Either NatsError ())
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> ClientState -> IO ConnectionState
readStatus ClientState
client'
            Maybe PingResult
Nothing -> do
              ConnectionState
status <- ClientState -> IO ConnectionState
readStatus ClientState
client'
              Either NatsError () -> IO (Either NatsError ())
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (NatsError -> Either NatsError ()
forall a b. a -> Either a b
Left (ConnectionState -> NatsError
closedError ConnectionState
status))

runningResult :: ClientState -> IO (Either NatsError ())
runningResult :: ClientState -> IO (Either NatsError ())
runningResult ClientState
client = do
  Either ClientExitReason ()
result <- STM (Either ClientExitReason ()) -> IO (Either ClientExitReason ())
forall a. STM a -> IO a
atomically (ClientState -> STM (Either ClientExitReason ())
waitForConnected ClientState
client)
  Either NatsError () -> IO (Either NatsError ())
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Either NatsError () -> IO (Either NatsError ()))
-> Either NatsError () -> IO (Either NatsError ())
forall a b. (a -> b) -> a -> b
$
    case Either ClientExitReason ()
result of
      Left ClientExitReason
reason -> NatsError -> Either NatsError ()
forall a b. a -> Either a b
Left (ClientExitReason -> NatsError
NatsConnectionClosed ClientExitReason
reason)
      Right ()    -> () -> Either NatsError ()
forall a b. b -> Either a b
Right ()

closedError :: ConnectionState -> NatsError
closedError :: ConnectionState -> NatsError
closedError ConnectionState
status =
  case ConnectionState
status of
    ConnectionClosing ClientExitReason
reason -> ClientExitReason -> NatsError
NatsConnectionClosed ClientExitReason
reason
    ConnectionClosed ClientExitReason
reason  -> ClientExitReason -> NatsError
NatsConnectionClosed ClientExitReason
reason
    ConnectionState
_                        -> ClientExitReason -> NatsError
NatsConnectionClosed ClientExitReason
ExitResetRequested