API.hs

-- | Internal home for the core NATS client capability surface.
--
-- The public @API@ module deliberately hides the constructors defined here.
-- Keeping construction private lets the capability grow without changing the
-- public representation.
module Client.API
  ( Client (..)
  , Message (..)
  , MsgView
  , Subscription (..)
  , subscriptionSid
  , NatsError (..)
  , Subject
  , Payload
  , Headers
  , PublishOption
  , SubscribeOption
  , RequestOption
  , UnsubscribeOption
  , PingOption
  , FlushOption
  , ResetOption
  , CloseOption
  , RequestConfig (..)
  , UnsubscribeConfig (..)
  , PingConfig (..)
  , FlushConfig (..)
  , ResetConfig (..)
  , CloseConfig (..)
  , withSubscriptionExpiry
  , withQueueGroup
  , withReplyTo
  , withHeaders
  , withRequestTimeout
  , withRequestHeaders
  ) where

import qualified Data.ByteString    as BS
import           Data.Time.Clock    (NominalDiffTime)
import           Lib.CallOption     (CallOption)
import           Publish.Config     (PublishConfig (..))
import           State.Types        (ClientExitReason)
import           Subscription.Types (SubscribeConfig (..))
import           Types.Msg          (Headers, Payload, SID, Subject)

-- | Client capabilities for publishing, subscribing, request-reply, and
-- lifecycle control.
--
-- This constructor is internal. Public callers receive a 'Client' from
-- 'Client.newClient' and use the named operations re-exported by @API@.
data Client = Client
                { publish :: Subject -> Payload -> [PublishOption] -> IO (Either NatsError ())
                  -- ^ Publish a message.
                , subscribe :: Subject -> [SubscribeOption] -> (Message -> IO ()) -> IO (Either NatsError Subscription)
                  -- ^ Subscribe to a subject until explicitly unsubscribed.
                , subscribeOnce :: Subject -> [SubscribeOption] -> (Message -> IO ()) -> IO (Either NatsError Subscription)
                  -- ^ Subscribe until the first message is delivered.
                , request :: Subject -> Payload -> [RequestOption] -> IO (Either NatsError Message)
                  -- ^ Publish a request and wait for one response.
                , unsubscribe :: Subscription -> [UnsubscribeOption] -> IO (Either NatsError ())
                  -- ^ Unsubscribe a subscription created by this client.
                , newInbox :: IO Subject
                  -- ^ Create a unique inbox subject for replies.
                , ping :: [PingOption] -> IO (Either NatsError ())
                  -- ^ Round trip a PING through the server.
                , flush :: [FlushOption] -> IO (Either NatsError ())
                  -- ^ Wait until all previously buffered writes reach the server.
                , reset :: [ResetOption] -> IO ()
                  -- ^ Reset the client connection state.
                , close :: [CloseOption] -> IO ()
                -- ^ Close the client connection and release resources.
                }

-- | A message delivered by NATS.
--
-- NATS always has a payload, which may be empty. This is intentionally
-- represented by a strict 'BS.ByteString', not @Maybe ByteString@.
data Message = Message
                 { subject :: Subject
                 , sid     :: SID
                 , replyTo :: Maybe Subject
                 , payload :: Payload
                 , headers :: Maybe Headers
                 }
  deriving (Eq, Show)

-- | Compatibility name retained for code written against pre-0.4 releases.
type MsgView = Message

-- | An opaque handle to an active subscription.
newtype Subscription = Subscription SID
  deriving (Eq, Show)

-- | The protocol identifier used by the legacy @API@ module. New code should
-- treat 'Subscription' as an opaque handle.
subscriptionSid :: Subscription -> SID
subscriptionSid (Subscription value) = value

-- | Failures reported by core NATS operations.
data NatsError = NatsValidationError BS.ByteString
               | NatsPayloadTooLarge Int Int
               | NatsSlowConsumer
               | NatsConnectionClosed ClientExitReason
               | NatsRequestTimedOut
               | NatsNoResponders
  deriving (Eq, Show)

type PublishOption = CallOption PublishConfig

type SubscribeOption = CallOption SubscribeConfig

data RequestConfig = RequestConfig
                       { requestTimeout :: NominalDiffTime
                       , requestHeaders :: Maybe Headers
                       }

type RequestOption = CallOption RequestConfig

data UnsubscribeConfig = UnsubscribeConfig
type UnsubscribeOption = CallOption UnsubscribeConfig

data PingConfig = PingConfig
type PingOption = CallOption PingConfig

data FlushConfig = FlushConfig
type FlushOption = CallOption FlushConfig

data ResetConfig = ResetConfig
type ResetOption = CallOption ResetConfig

data CloseConfig = CloseConfig
type CloseOption = CallOption CloseConfig

-- | Stop a one-shot subscription if it has not received a message within the
-- given interval.
withSubscriptionExpiry :: NominalDiffTime -> SubscribeOption
withSubscriptionExpiry expirySeconds cfg = cfg { expiry = Just expirySeconds }

-- | Deliver messages to one member of the named queue group.
withQueueGroup :: Subject -> SubscribeOption
withQueueGroup queueGroup cfg = cfg { subscribeQueueGroup = Just queueGroup }

-- | Set the reply subject on an outgoing publish.
withReplyTo :: Subject -> PublishOption
withReplyTo replySubject cfg = cfg { publishReplyTo = Just replySubject }

-- | Set headers on an outgoing publish.
withHeaders :: Headers -> PublishOption
withHeaders messageHeaders cfg = cfg { publishHeaders = Just messageHeaders }

-- | Set the amount of time to wait for a request reply.
withRequestTimeout :: NominalDiffTime -> RequestOption
withRequestTimeout timeout cfg = cfg { requestTimeout = max 0 timeout }

-- | Set headers on an outgoing request.
withRequestHeaders :: Headers -> RequestOption
withRequestHeaders messageHeaders cfg = cfg { requestHeaders = Just messageHeaders }