{-# LANGUAGE OverloadedStrings #-} module JetStream.Protocol.Request ( requestMsg , requestJSON , requestJSONWithHeaders , requestUnit , decodeJetStreamResponse ) where import qualified Client.API as Nats import Control.Monad (void) import Data.Aeson import qualified Data.Aeson as Aeson import Data.Aeson.Types (Parser, parseEither) import qualified Data.ByteString as BS import qualified Data.ByteString.Lazy as LBS import JetStream.Error import JetStream.Options ( JetStreamContext (..) , requestTimeoutMicros ) import JetStream.Protocol.Headers (statusError) import JetStream.Types ( Headers , JetStreamRequestOption , Payload , Subject ) requestMsg :: JetStreamContext -> Subject -> Payload -> Headers -> [JetStreamRequestOption] -> IO (Either JetStreamError Nats.MsgView) requestMsg ctx subject payload headers options = do result <- Nats.request (contextClient ctx) subject payload requestOptions pure $ case result of Left Nats.NatsRequestTimedOut -> Left JetStreamTimeout Left err -> Left (JetStreamNatsError err) Right msg -> Right msg where timeoutSeconds = fromRational (toRational (requestTimeoutMicros ctx options) / 1000000) requestOptions = Nats.withRequestTimeout timeoutSeconds : [Nats.withRequestHeaders headers | not (null headers)] requestJSON :: FromJSON a => JetStreamContext -> Subject -> Maybe Value -> [JetStreamRequestOption] -> IO (Either JetStreamError a) requestJSON ctx subject body = requestJSONWithHeaders ctx subject body [] requestJSONWithHeaders :: FromJSON a => JetStreamContext -> Subject -> Maybe Value -> Headers -> [JetStreamRequestOption] -> IO (Either JetStreamError a) requestJSONWithHeaders ctx subject body headers options = do msgResult <- requestMsg ctx subject (maybe BS.empty strictEncode body) headers options pure $ do msg <- msgResult case statusError msg of Just err -> Left err Nothing -> decodeJetStreamResponse (Nats.payload msg) requestUnit :: JetStreamContext -> Subject -> Maybe Value -> [JetStreamRequestOption] -> IO (Either JetStreamError ()) requestUnit ctx subject body options = do response <- requestJSON ctx subject body options pure (void (response :: Either JetStreamError SuccessResponse)) decodeJetStreamResponse :: FromJSON a => BS.ByteString -> Either JetStreamError a decodeJetStreamResponse payload = case Aeson.eitherDecode (LBS.fromStrict payload) :: Either String Value of Left err -> Left (JetStreamDecodeError err) Right value -> case parseEither parseApiResult value of Left err -> Left (JetStreamDecodeError err) Right result -> result parseApiResult :: FromJSON a => Value -> Parser (Either JetStreamError a) parseApiResult value = case value of Object obj -> do apiError <- obj .:? "error" case apiError of Just err -> pure (Left (JetStreamApiFailure err)) Nothing -> Right <$> parseJSON value _ -> Right <$> parseJSON value strictEncode :: Value -> BS.ByteString strictEncode = LBS.toStrict . Aeson.encode newtype SuccessResponse = SuccessResponse Bool deriving (Eq, Show) instance FromJSON SuccessResponse where parseJSON = withObject "SuccessResponse" $ \obj -> SuccessResponse <$> obj .: "success"