module JetStream.Stream ( streamAPI , module JetStream.Stream.API ) where import Data.Aeson (toJSON) import JetStream.Error ( JetStreamApiError (..) , JetStreamError (..) ) import JetStream.Options (JetStreamContext) import qualified JetStream.Protocol.Request as Request import qualified JetStream.Protocol.Subject as Subject import JetStream.Stream.API import JetStream.Stream.Types ( StreamAPI (..) , purgeStreamRequest , streamConfigRequest , streamListRequest , streamMessageDeleteRequest , streamMessageGetRequest , streamNamesRequest ) streamAPI :: JetStreamContext -> StreamAPI streamAPI context = StreamAPI { create = \name subjects options requestOptions -> let config = streamConfigRequest name subjects options in Request.requestJSON context (Subject.streamCreateSubject context name) (Just (toJSON config)) requestOptions , createOrUpdate = \name subjects options requestOptions -> do updateResult <- update (streamAPI context) name subjects options requestOptions case updateResult of Left err | isStreamNotFound err -> create (streamAPI context) name subjects options requestOptions other -> pure other , update = \name subjects options requestOptions -> let config = streamConfigRequest name subjects options in Request.requestJSON context (Subject.streamUpdateSubject context name) (Just (toJSON config)) requestOptions , info = \streamName requestOptions -> Request.requestJSON context (Subject.streamInfoSubject context streamName) Nothing requestOptions , getMessage = \streamName selector requestOptions -> Request.requestJSON context (Subject.streamMessageGetSubject context streamName) (Just (toJSON (streamMessageGetRequest selector))) requestOptions , deleteMessage = \streamName sequenceNumber mode requestOptions -> Request.requestJSON context (Subject.streamMessageDeleteSubject context streamName) (Just (toJSON (streamMessageDeleteRequest sequenceNumber mode))) requestOptions , delete = \streamName requestOptions -> Request.requestJSON context (Subject.streamDeleteSubject context streamName) Nothing requestOptions , purge = \streamName options requestOptions -> Request.requestJSON context (Subject.streamPurgeSubject context streamName) (Just (toJSON (purgeStreamRequest options))) requestOptions , list = Request.requestJSON context (Subject.streamListSubject context) . Just . toJSON . streamListRequest , names = Request.requestJSON context (Subject.streamNamesSubject context) . Just . toJSON . streamNamesRequest } isStreamNotFound :: JetStreamError -> Bool isStreamNotFound (JetStreamApiFailure err) = apiErrorCodeDetail err == 10059 isStreamNotFound _ = False