{-# LANGUAGE OverloadedStrings #-}

-- | High-level client implementation for NATS.
module Client
  ( Server
  , ServerConfigError (..)
  , server
  , serverWithDefaultPort
  , serverHost
  , serverPort
  , connect
  , newClient
  , ConfigOption
  , withConnectName
  , withEcho
  , withAuthToken
  , withAuthTokenHandler
  , withUserPass
  , withUserPassHandler
  , withNKey
  , withNKeyHandler
  , withJWT
  , withJWTHandlers
  , withTLS
  , withTLSCert
  , withTLSRootCA
  , withTLSServerName
  , withTLSInsecure
  , withMinimumLogLevel
  , withLogAction
  , withConnectionAttempts
  , withConnectTimeoutMicros
  , withCallbackConcurrency
  , withMessageLimit
  , withPendingDeliveryLimits
  , withErrorHandler
  , withBufferLimit
  , withExitAction
  , LogLevel (..)
  , LogEntry (..)
  , renderLogEntry
  , AuthTokenData
  , AuthTokenHandler
  , UserPassData
  , UserPassHandler
  , NKeyData
  , NKeyPublicKey
  , JWTTokenData
  , JWTHandler
  , SignatureHandler
  , AuthError (..)
  , TLSPublicKey
  , TLSPrivateKey
  , TLSCertData
  , TLSConfig (..)
  , ClientExitReason (..)
  , ServerError
  , serverErrorReason
  , 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 (..)
    )
import           Control.Concurrent       (forkIO)
import           Control.Concurrent.STM
import           Control.Exception
    ( SomeException
    , displayException
    , 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)
import           Data.Time.Clock          (NominalDiffTime)
import           Data.Version             (showVersion)
import           Engine                   (closeClient, resetClient, runEngine)
import           Lib.CallOption           (CallOption, applyCallOptions)
import           Lib.Logger
    ( LogEntry (..)
    , LogLevel (..)
    , LoggerConfig (..)
    , MonadLogger (..)
    , defaultLogger
    , newLogContext
    , renderLogEntry
    )
import           Network.Connection       (connectionApi)
import           Network.ConnectionAPI    (newConn)
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 (QueueItem))
import           Queue.TransactionalQueue (newQueue)
import           State.Store
    ( ClientState
    , config
    , enqueue
    , newClientState
    , nextInbox
    , nextSid
    , pushPingAction
    , readServerInfo
    , readStatus
    , runClient
    , setConnectName
    , waitForClosed
    , waitForInitialConnection
    , waitForNotRunning
    )
import           State.Types
    ( ClientConfig (..)
    , ClientExitReason (..)
    , ClientStatus (..)
    , ConnectAttemptError (..)
    , ConnectError (..)
    , ConnectFailure (..)
    , ServerError
    , TLSCertData
    , TLSConfig (..)
    , TLSPrivateKey
    , TLSPublicKey
    , defaultTLSConfig
    , serverErrorReason
    )
import           Subscription.Store
    ( SubscriptionStore
    , awaitNoTrackedExpiries
    , hasTrackedExpiries
    , newSubscriptionStore
    , register
    , startExpiryWorker
    , startWorkers
    , unregister
    )
import           Subscription.Types
    ( PendingLimits (..)
    , SubscribeConfig (..)
    , SubscriptionMeta (SubscriptionMeta)
    , defaultPendingLimits
    )
import qualified Types.Connect            as Connect
import qualified Types.Info               as Info
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 -> 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 -> ClientExitReason -> IO ()
optionExitAction           :: ClientExitReason -> IO ()
                       , ClientOptions -> [([Char], Int)]
optionConnectOptions       :: [(String, Int)]
                       }

-- | 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 = 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
        , 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 ())
        , 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
          , 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

  Int
-> SubscriptionStore -> STM () -> (SomeException -> IO ()) -> IO ()
startWorkers
    (ClientConfig -> Int
callbackConcurrency ClientConfig
clientConfig)
    SubscriptionStore
store
    (do
        ClientState -> STM ()
waitForClosed ClientState
clientState
        SubscriptionStore -> STM ()
awaitNoTrackedExpiries SubscriptionStore
store)
    (ClientState -> SomeException -> IO ()
handleCallbackError ClientState
clientState)

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

  IO ThreadId -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO ThreadId -> IO ()) -> (IO () -> IO ThreadId) -> IO () -> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. IO () -> IO ThreadId
forkIO (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$
    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

  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
-> SID -> SID -> PublishConfig -> IO (Either NatsError ())
publishClient ClientState
clientState 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
-> Bool
-> SID
-> SubscribeConfig
-> (Message -> IO ())
-> IO (Either NatsError Subscription)
subscribeClient ClientState
clientState SubscriptionStore
store Bool
False 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
-> Bool
-> SID
-> SubscribeConfig
-> (Message -> IO ())
-> IO (Either NatsError Subscription)
subscribeClient ClientState
clientState SubscriptionStore
store Bool
True 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
        , 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
PingConfig of
              PingConfig
PingConfig -> ClientState -> IO (Either NatsError ())
flushClient ClientState
clientState
        , flush :: [FlushOption] -> IO (Either NatsError ())
flush = \[FlushOption]
options ->
            case [FlushOption] -> FlushOption
forall a. [CallOption a] -> CallOption a
applyCallOptions [FlushOption]
options FlushConfig
FlushConfig of
              FlushConfig
FlushConfig -> ClientState -> IO (Either NatsError ())
flushClient 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 -> ConnectionAPI -> ClientState -> SubscriptionStore -> IO ()
closeClient ConnectionAPI
connectionApi ClientState
clientState SubscriptionStore
store
        }

  Either ConnectError ()
initialResult <- STM (Either ConnectError ()) -> IO (Either ConnectError ())
forall a. STM a -> IO a
atomically (ClientState -> STM (Either ConnectError ())
waitForInitialConnection ClientState
clientState)
  Either ConnectError Client -> IO (Either ConnectError Client)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Client
client Client -> Either ConnectError () -> Either ConnectError Client
forall a b. a -> Either ConnectError b -> Either ConnectError a
forall (f :: * -> *) a b. Functor f => a -> f b -> f a
<$ Either ConnectError ()
initialResult)

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 }

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

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 maximum time for INFO, TLS, CONNECT, and PONG negotiation 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 }

-- | 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

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 of
      Maybe TLSConfig
Nothing ->
        () -> AppM ()
forall a. a -> AppM a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
      Just TLSConfig
tls -> 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
err =
  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: " [Char] -> ShowS
forall a. [a] -> [a] -> [a]
++ SomeException -> [Char]
forall e. Exception e => e -> [Char]
displayException SomeException
err)

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
  ClientStatus
status <- ClientState -> IO ClientStatus
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 ClientStatus
status of
      Closed ClientExitReason
_ -> Bool -> Bool
not Bool
tracked
      ClientStatus
_        -> 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
  -> Msg.Subject
  -> Msg.Payload
  -> PublishConfig
  -> IO (Either NatsError ())
publishClient :: ClientState
-> SID -> SID -> PublishConfig -> IO (Either NatsError ())
publishClient ClientState
client 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
          }
  Either NatsError ()
validation <- ClientState -> Pub -> IO (Either NatsError ())
canPublish 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
      ClientState -> QueueItem -> IO ()
enqueue ClientState
client (Pub -> QueueItem
forall m. Transformer m => m -> QueueItem
QueueItem Pub
publishMessage)
      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 ())

canPublish :: ClientState -> Pub.Pub -> IO (Either NatsError ())
canPublish :: ClientState -> Pub -> IO (Either NatsError ())
canPublish 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
      Either NatsError ()
statusResult <- ClientState -> IO (Either NatsError ())
runningResult ClientState
client
      case Either NatsError ()
statusResult 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
          Maybe Info
serverInfo <- ClientState -> IO (Maybe Info)
readServerInfo ClientState
client
          let actual :: Int
actual = Pub -> Int
Pub.messageSize Pub
publishMessage
              clientMaximum :: Int
clientMaximum = ClientConfig -> Int
messageLimit (ClientState -> ClientConfig
config ClientState
client)
              maximumSize :: Int
maximumSize =
                Int -> (Info -> Int) -> Maybe Info -> Int
forall b a. b -> (a -> b) -> Maybe a -> b
maybe
                  Int
clientMaximum
                  (Int -> Int -> Int
forall a. Ord a => a -> a -> a
min Int
clientMaximum (Int -> Int) -> (Info -> Int) -> Info -> Int
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Info -> Int
Info.max_payload)
                  Maybe Info
serverInfo
          if Int
actual Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
> Int
maximumSize
            then 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
actual
                      [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 () -> 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 (Int -> Int -> NatsError
NatsPayloadTooLarge Int
actual Int
maximumSize))
            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 ())

subscribeClient
  :: ClientState
  -> SubscriptionStore
  -> Bool
  -> Msg.Subject
  -> SubscribeConfig
  -> (Message -> IO ())
  -> IO (Either NatsError Subscription)
subscribeClient :: ClientState
-> SubscriptionStore
-> Bool
-> SID
-> SubscribeConfig
-> (Message -> IO ())
-> IO (Either NatsError Subscription)
subscribeClient ClientState
client SubscriptionStore
store Bool
isReply SID
subject SubscribeConfig
cfg Message -> IO ()
callback =
  ClientState
-> SubscriptionStore
-> Bool
-> SID
-> SubscribeConfig
-> (Maybe Msg -> IO ())
-> IO (Either NatsError Subscription)
subscribeRawClient ClientState
client SubscriptionStore
store Bool
isReply 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
  -> Bool
  -> Msg.Subject
  -> SubscribeConfig
  -> (Maybe Msg.Msg -> IO ())
  -> IO (Either NatsError Subscription)
subscribeRawClient :: ClientState
-> SubscriptionStore
-> Bool
-> SID
-> SubscribeConfig
-> (Maybe Msg -> IO ())
-> IO (Either NatsError Subscription)
subscribeRawClient ClientState
client SubscriptionStore
store Bool
isReply SID
subject SubscribeConfig
cfg Maybe Msg -> IO ()
callback = do
  ClientState
-> SubscriptionStore
-> Bool
-> SID
-> SubscribeConfig
-> (Maybe Msg -> IO ())
-> IO ()
-> IO (Either NatsError Subscription)
subscribeRawClientWithOverflow
    ClientState
client
    SubscriptionStore
store
    Bool
isReply
    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
  -> Bool
  -> Msg.Subject
  -> SubscribeConfig
  -> (Maybe Msg.Msg -> IO ())
  -> IO ()
  -> IO (Either NatsError Subscription)
subscribeRawClientWithOverflow :: ClientState
-> SubscriptionStore
-> Bool
-> SID
-> SubscribeConfig
-> (Maybe Msg -> IO ())
-> IO ()
-> IO (Either NatsError Subscription)
subscribeRawClientWithOverflow ClientState
client SubscriptionStore
store Bool
isReply SID
subject SubscribeConfig
cfg Maybe Msg -> IO ()
callback IO ()
onDropped = 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)
  Either NatsError ()
statusResult <- ClientState -> IO (Either NatsError ())
runningResult ClientState
client
  let queueGroup :: Maybe SID
queueGroup = SubscribeConfig -> Maybe SID
subscribeQueueGroup SubscribeConfig
cfg
  SID
sid <- ClientState -> IO SID
nextSid ClientState
client
  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
          }
  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 () -> case Either NatsError ()
statusResult of
      Left NatsError
err -> 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 () -> do
        let meta :: SubscriptionMeta
meta =
              SID -> Maybe SID -> Bool -> SubscriptionMeta
SubscriptionMeta SID
subject Maybe SID
queueGroup Bool
isReply
        SubscriptionStore
-> SID
-> SubscriptionMeta
-> SubscribeConfig
-> (Maybe Msg -> IO ())
-> IO ()
-> IO ()
register SubscriptionStore
store SID
sid SubscriptionMeta
meta SubscribeConfig
cfg Maybe Msg -> IO ()
callback IO ()
onDropped
        ClientState -> QueueItem -> IO ()
enqueue ClientState
client (QueueItem -> IO ()) -> QueueItem -> IO ()
forall a b. (a -> b) -> a -> b
$
          Sub -> QueueItem
forall m. Transformer m => m -> QueueItem
QueueItem
            Sub
subscriptionMessage
        Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
isReply (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$
          ClientState -> QueueItem -> IO ()
enqueue ClientState
client
            (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
                })
        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 (SID -> Subscription
Subscription SID
sid))

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 =
  ((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
    TVar Bool
deadline <- Int -> IO (TVar Bool)
registerDelay (NominalDiffTime -> Int
durationMicros (RequestConfig -> NominalDiffTime
requestTimeout RequestConfig
cfg))
    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 :: IO ()
rejectSlowConsumer =
          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
NatsSlowConsumer)))
    Either NatsError Subscription
subscriptionResult <-
      ClientState
-> SubscriptionStore
-> Bool
-> SID
-> SubscribeConfig
-> (Maybe Msg -> IO ())
-> IO ()
-> IO (Either NatsError Subscription)
subscribeRawClientWithOverflow
        ClientState
client
        SubscriptionStore
store
        Bool
True
        SID
inbox
        SubscribeConfig
subscriptionConfig
        Maybe Msg -> IO ()
deliver
        IO ()
rejectSlowConsumer
    case Either NatsError Subscription
subscriptionResult 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 Subscription
subscription -> do
        let publishConfig :: PublishConfig
publishConfig = Maybe Headers -> Maybe SID -> PublishConfig
PublishConfig (RequestConfig -> Maybe Headers
requestHeaders RequestConfig
cfg) (SID -> Maybe SID
forall a. a -> Maybe a
Just SID
inbox)
            cleanup :: IO ()
cleanup = IO (Either NatsError ()) -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (ClientState
-> SubscriptionStore -> Subscription -> IO (Either NatsError ())
unsubscribeClient ClientState
client SubscriptionStore
store Subscription
subscription)
        Either NatsError ()
publishResult <- ClientState
-> SID -> SID -> PublishConfig -> IO (Either NatsError ())
publishClient ClientState
client SID
requestSubject SID
requestPayload PublishConfig
publishConfig
        case Either NatsError ()
publishResult of
          Left NatsError
err -> IO ()
cleanup IO ()
-> IO (Either NatsError Message) -> IO (Either NatsError Message)
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> 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
            Either NatsError Message
result <- IO (Either NatsError Message) -> IO (Either NatsError Message)
forall a. IO a -> IO a
restore (ClientState
-> TVar Bool
-> TMVar (Either NatsError Message)
-> IO (Either NatsError Message)
awaitRequest ClientState
client TVar Bool
deadline 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
  -> TVar Bool
  -> TMVar (Either NatsError Message)
  -> IO (Either NatsError Message)
awaitRequest :: ClientState
-> TVar Bool
-> TMVar (Either NatsError Message)
-> IO (Either NatsError Message)
awaitRequest ClientState
client TVar Bool
deadline TMVar (Either NatsError Message)
response = do
  Maybe (Either NatsError Message)
outcome <- STM (Maybe (Either NatsError Message))
-> IO (Maybe (Either NatsError Message))
forall a. STM a -> IO a
atomically (STM (Maybe (Either NatsError Message))
 -> IO (Maybe (Either NatsError Message)))
-> STM (Maybe (Either NatsError Message))
-> IO (Maybe (Either NatsError Message))
forall a b. (a -> b) -> a -> b
$
    (Either NatsError Message -> Maybe (Either NatsError Message)
forall a. a -> Maybe a
Just (Either NatsError Message -> Maybe (Either NatsError Message))
-> STM (Either NatsError Message)
-> STM (Maybe (Either NatsError Message))
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> TMVar (Either NatsError Message) -> STM (Either NatsError Message)
forall a. TMVar a -> STM a
readTMVar TMVar (Either NatsError Message)
response)
      STM (Maybe (Either NatsError Message))
-> STM (Maybe (Either NatsError Message))
-> STM (Maybe (Either NatsError Message))
forall a. STM a -> STM a -> STM a
`orElse` (do
        Bool
expired <- TVar Bool -> STM Bool
forall a. TVar a -> STM a
readTVar TVar Bool
deadline
        Bool -> STM ()
check Bool
expired
        Maybe (Either NatsError Message)
-> STM (Maybe (Either NatsError Message))
forall a. a -> STM a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Either NatsError Message -> Maybe (Either NatsError Message)
forall a. a -> Maybe a
Just (NatsError -> Either NatsError Message
forall a b. a -> Either a b
Left NatsError
NatsRequestTimedOut)))
      STM (Maybe (Either NatsError Message))
-> STM (Maybe (Either NatsError Message))
-> STM (Maybe (Either NatsError Message))
forall a. STM a -> STM a -> STM a
`orElse` (Maybe (Either NatsError Message)
forall a. Maybe a
Nothing Maybe (Either NatsError Message)
-> STM () -> STM (Maybe (Either NatsError Message))
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 (Either NatsError Message)
outcome of
    Just Either NatsError Message
result -> 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
    Maybe (Either NatsError Message)
Nothing     -> do
      ClientStatus
status <- ClientState -> IO ClientStatus
readStatus ClientState
client
      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 (ClientStatus -> NatsError
closedError ClientStatus
status))

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 (Double -> Integer
forall b. Integral b => Double -> b
forall a b. (RealFrac a, Integral b) => a -> b
floor (NominalDiffTime -> Double
forall a b. (Real a, Fractional b) => a -> b
realToFrac NominalDiffTime
duration Double -> Double -> Double
forall a. Num a => a -> a -> a
* (Double
1000000 :: Double)))

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) = 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)
  SubscriptionStore -> SID -> IO ()
unregister SubscriptionStore
store SID
sid
  Either NatsError ()
statusResult <- ClientState -> IO (Either NatsError ())
runningResult ClientState
client
  case Either NatsError ()
statusResult 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
      ClientState -> QueueItem -> IO ()
enqueue ClientState
client (QueueItem -> IO ()) -> QueueItem -> IO ()
forall a b. (a -> b) -> a -> b
$
        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
            }
      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 ())

pingClient :: ClientState -> IO () -> IO ()
pingClient :: ClientState -> IO () -> IO ()
pingClient ClientState
client IO ()
action = 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]
"sending ping to server"
  ClientState -> IO () -> IO ()
pushPingAction ClientState
client IO ()
action
  ClientState -> QueueItem -> IO ()
enqueue ClientState
client (Ping -> QueueItem
forall m. Transformer m => m -> QueueItem
QueueItem Ping
Ping)

flushClient :: ClientState -> IO (Either NatsError ())
flushClient :: ClientState -> IO (Either NatsError ())
flushClient ClientState
client = do
  Either NatsError ()
statusResult <- ClientState -> IO (Either NatsError ())
runningResult ClientState
client
  case Either NatsError ()
statusResult 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
      TMVar ()
ponged <- IO (TMVar ())
forall a. IO (TMVar a)
newEmptyTMVarIO
      ClientState -> IO () -> IO ()
pingClient ClientState
client (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 ()
ponged ())))
      Bool
outcome <- STM Bool -> IO Bool
forall a. STM a -> IO a
atomically (STM Bool -> IO Bool) -> STM Bool -> IO Bool
forall a b. (a -> b) -> a -> b
$
        (Bool
True Bool -> STM () -> STM Bool
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 ()
ponged)
          STM Bool -> STM Bool -> STM Bool
forall a. STM a -> STM a -> STM a
`orElse` (Bool
False Bool -> STM () -> STM Bool
forall a b. a -> STM b -> STM a
forall (f :: * -> *) a b. Functor f => a -> f b -> f a
<$ ClientState -> STM ()
waitForNotRunning ClientState
client)
      if Bool
outcome
        then 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 ())
        else do
          ClientStatus
status <- ClientState -> IO ClientStatus
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 (ClientStatus -> NatsError
closedError ClientStatus
status))

runningResult :: ClientState -> IO (Either NatsError ())
runningResult :: ClientState -> IO (Either NatsError ())
runningResult ClientState
client = do
  ClientStatus
status <- ClientState -> IO ClientStatus
readStatus 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 ClientStatus
status of
      ClientStatus
Running        -> () -> Either NatsError ()
forall a b. b -> Either a b
Right ()
      Closing ClientExitReason
reason -> NatsError -> Either NatsError ()
forall a b. a -> Either a b
Left (ClientExitReason -> NatsError
NatsConnectionClosed ClientExitReason
reason)
      Closed ClientExitReason
reason  -> NatsError -> Either NatsError ()
forall a b. a -> Either a b
Left (ClientExitReason -> NatsError
NatsConnectionClosed ClientExitReason
reason)

closedError :: ClientStatus -> NatsError
closedError :: ClientStatus -> NatsError
closedError ClientStatus
status =
  case ClientStatus
status of
    Closing ClientExitReason
reason -> ClientExitReason -> NatsError
NatsConnectionClosed ClientExitReason
reason
    Closed ClientExitReason
reason  -> ClientExitReason -> NatsError
NatsConnectionClosed ClientExitReason
reason
    ClientStatus
Running        -> ClientExitReason -> NatsError
NatsConnectionClosed ClientExitReason
ExitResetRequested