natskell-1.1.0.0: test/Unit/JetStream/ConsumerSpec.hs
{-# LANGUAGE OverloadedStrings #-}
module JetStream.ConsumerSpec (spec) where
import Data.Aeson
( Value
, eitherDecode
, encode
, object
, (.=)
)
import qualified Data.ByteString.Lazy as LBS
import Data.Int (Int64)
import Data.Ratio ((%))
import JetStream.Consumer.Types
import JetStream.Error
( JetStreamError (JetStreamDecodeError)
)
import Test.Hspec
spec :: Spec
spec = do
describe "ConsumerConfig request JSON" $ do
it "encodes only populated fields" $ do
eitherDecode (encode durablePullConsumerConfigRequest) `shouldBe` Right durablePullConsumerConfigValue
it "encodes push and ordered consumer fields" $ do
eitherDecode (encode pushConsumerConfigRequest) `shouldBe` Right pushConsumerConfigValue
it "lets consumer target and kind own mutually exclusive fields" $ do
eitherDecode (encode targetKindConsumerConfigRequest)
`shouldBe` Right targetKindConsumerConfigValue
it "applies consumer options from left to right" $ do
let request = consumerConfigRequest
[ withConsumerDeliverPolicy DeliverAll
, withConsumerDeliverPolicy DeliverLast
]
eitherDecode (encode request)
`shouldBe` Right (object ["deliver_policy" .= ("last" :: String)])
it "encodes backoff durations as exact integer nanoseconds" $ do
let request = consumerConfigRequest
[ withConsumerBackoff
[ fromRational (1234567891 % 1000000000)
, fromRational (1 % 1000000000)
]
, withConsumerMaxRequestBatch maxBound
]
validateConsumerConfigRequest request `shouldBe` Right ()
eitherDecode (encode request) `shouldBe` Right (object
[ "backoff" .= [1234567891 :: Integer, 1]
, "max_batch" .= (maxBound :: Int)
])
it "floors backoff at sub-nanosecond and Int64 boundaries" $ do
let request = consumerConfigRequest
[withConsumerBackoff [subNanosecond, oneNanosecond, maxInt64Nanoseconds]]
validateConsumerConfigRequest request `shouldBe` Right ()
eitherDecode (encode request) `shouldBe` Right (object
["backoff" .= [0 :: Integer, 1, toInteger (maxBound :: Int64)]])
it "rejects negative and overflowing backoff durations" $ do
validateConsumerConfigRequest
(consumerConfigRequest [withConsumerBackoff [negativeSubNanosecond]])
`shouldBe` Left (JetStreamDecodeError backoffRangeError)
validateConsumerConfigRequest
(consumerConfigRequest [withConsumerBackoff [overflowingNanoseconds]])
`shouldBe` Left (JetStreamDecodeError backoffRangeError)
it "rejects negative max request batch" $ do
validateConsumerConfigRequest
(consumerConfigRequest [withConsumerMaxRequestBatch (-1)])
`shouldBe` Left (JetStreamDecodeError "consumer max request batch must be zero or greater")
it "allows backoff no longer than positive max deliver" $ do
validateConsumerConfigRequest (backoffWithMaxDeliver 2 [1]) `shouldBe` Right ()
validateConsumerConfigRequest (backoffWithMaxDeliver 2 [1, 2]) `shouldBe` Right ()
validateConsumerConfigRequest (backoffWithMaxDeliver 0 [1, 2, 3]) `shouldBe` Right ()
validateConsumerConfigRequest (backoffWithMaxDeliver (-1) [1, 2, 3]) `shouldBe` Right ()
validateConsumerConfigRequest (backoffWithMaxDeliver 2 [1, 2, 3])
`shouldBe` Left (JetStreamDecodeError
"consumer backoff length cannot exceed positive max deliver")
it "omits empty backoff and zero max request batch values" $ do
let request = consumerConfigRequest
[ withConsumerBackoff []
, withConsumerMaxRequestBatch 0
]
validateConsumerConfigRequest request `shouldBe` Right ()
eitherDecode (encode request) `shouldBe` Right (object [])
describe "Consumer reset request JSON" $ do
it "encodes an optional reset sequence" $ do
eitherDecode (encode (consumerResetRequest [withConsumerResetSequence 7]))
`shouldBe` Right (object ["seq" .= (7 :: Int)])
describe "ConsumerInfo JSON" .
it "decodes server responses into concrete fields" $ do
eitherDecode consumerInfoJSON `shouldBe` Right consumerInfoFixture
describe "ConsumerConfig response JSON" $ do
it "normalizes absent backoff and zero max request batch to unset" $ do
fmap configResourceLimits (eitherDecode zeroMaxBatchConsumerConfigJSON)
`shouldBe` Right (Nothing, Nothing)
it "round-trips backoff and max request batch" $ do
eitherDecode (encode (consumerInfoConfig consumerInfoFixture))
`shouldBe` Right (consumerInfoConfig consumerInfoFixture)
describe "ConsumerResetResponse JSON" .
it "decodes reset metadata with the updated consumer info" $ do
fmap consumerResetResponseSequence (eitherDecode consumerResetJSON)
`shouldBe` Right 7
describe "Consumer list JSON" $ do
it "decodes detailed consumer list responses" $ do
eitherDecode consumerListJSON
`shouldBe` Right ConsumerListResponse
{ consumerListTotal = 1
, consumerListOffset = 0
, consumerListLimit = 1024
, consumerListConsumers = [consumerInfoFixture]
}
it "decodes consumer name responses" $ do
eitherDecode consumerNamesJSON
`shouldBe` Right ConsumerNamesResponse
{ consumerNamesTotal = 2
, consumerNamesOffset = 0
, consumerNamesLimit = 1024
, consumerNamesConsumers = ["orders-puller", "orders-worker"]
}
configResourceLimits config =
(consumerConfigBackoff config, consumerConfigMaxRequestBatch config)
backoffWithMaxDeliver maxDeliver backoff =
consumerConfigRequest
[ withConsumerMaxDeliver maxDeliver
, withConsumerBackoff backoff
]
subNanosecond = fromRational (1 % 2000000000)
negativeSubNanosecond = fromRational ((-1) % 2000000000)
oneNanosecond = fromRational (1 % 1000000000)
maxInt64Nanoseconds =
fromRational (toInteger (maxBound :: Int64) % 1000000000)
overflowingNanoseconds =
fromRational ((toInteger (maxBound :: Int64) + 1) % 1000000000)
backoffRangeError =
"consumer backoff durations must floor to nanoseconds between 0 and 9223372036854775807"
durablePullConsumerConfigRequest :: ConsumerConfigRequest
durablePullConsumerConfigRequest =
consumerConfigRequest
[ withConsumerDurableName "orders-puller"
, withConsumerName "orders-puller"
, withConsumerDeliverPolicy DeliverAll
, withConsumerAckPolicy AckExplicit
, withConsumerReplayPolicy ReplayInstant
, withConsumerFilter (ConsumerFilterSubjects ["orders.created", "orders.updated"])
, withConsumerMaxDeliver 5
, withConsumerAckWait 30
, withConsumerMaxAckPending 256
, withConsumerInactiveThreshold 60
]
durablePullConsumerConfigValue :: Value
durablePullConsumerConfigValue = object
[ "durable_name" .= ("orders-puller" :: String)
, "name" .= ("orders-puller" :: String)
, "deliver_policy" .= ("all" :: String)
, "ack_policy" .= ("explicit" :: String)
, "replay_policy" .= ("instant" :: String)
, "filter_subjects" .= ["orders.created" :: String, "orders.updated"]
, "max_deliver" .= (5 :: Int)
, "ack_wait" .= (30000000000 :: Integer)
, "max_ack_pending" .= (256 :: Int)
, "inactive_threshold" .= (60000000000 :: Integer)
]
pushConsumerConfigRequest :: ConsumerConfigRequest
pushConsumerConfigRequest =
consumerConfigRequest
[ withConsumerName "push"
, withConsumerDeliverSubject "deliver"
, withConsumerDeliverGroup "workers"
, withConsumerAckPolicy AckExplicit
, withConsumerMaxWaiting 512
, withConsumerIdleHeartbeat 1
, withConsumerHeadersOnly True
, withConsumerReplicas 1
, withConsumerMemoryStorage True
]
pushConsumerConfigValue :: Value
pushConsumerConfigValue = object
[ "name" .= ("push" :: String)
, "deliver_subject" .= ("deliver" :: String)
, "deliver_group" .= ("workers" :: String)
, "ack_policy" .= ("explicit" :: String)
, "max_waiting" .= (512 :: Int)
, "idle_heartbeat" .= (1000000000 :: Integer)
, "headers_only" .= True
, "num_replicas" .= (1 :: Int)
, "mem_storage" .= True
]
targetKindConsumerConfigRequest :: ConsumerConfigRequest
targetKindConsumerConfigRequest =
applyConsumerTarget (NamedConsumer "right") .
applyConsumerKind (PushConsumer "deliver") $
consumerConfigRequest
[ withConsumerName "wrong"
, withConsumerDeliverSubject "wrong"
, withConsumerDeliverGroup "workers"
]
targetKindConsumerConfigValue :: Value
targetKindConsumerConfigValue = object
[ "name" .= ("right" :: String)
, "deliver_subject" .= ("deliver" :: String)
, "deliver_group" .= ("workers" :: String)
]
consumerInfoJSON :: LBS.ByteString
consumerInfoJSON =
"{\"type\":\"io.nats.jetstream.api.v1.consumer_info_response\",\"stream_name\":\"ORDERS\",\"name\":\"orders-puller\",\"created\":\"2024-01-01T00:00:00Z\",\"config\":{\"deliver_policy\":\"all\",\"ack_policy\":\"explicit\",\"replay_policy\":\"instant\",\"backoff\":[1000000001,2],\"max_batch\":128},\"delivered\":{\"consumer_seq\":2,\"stream_seq\":10},\"ack_floor\":{\"consumer_seq\":1,\"stream_seq\":9},\"num_ack_pending\":1,\"num_redelivered\":0,\"num_waiting\":0,\"num_pending\":3}"
zeroMaxBatchConsumerConfigJSON :: LBS.ByteString
zeroMaxBatchConsumerConfigJSON =
"{\"deliver_policy\":\"all\",\"ack_policy\":\"explicit\",\"replay_policy\":\"instant\",\"max_batch\":0}"
consumerResetJSON :: LBS.ByteString
consumerResetJSON =
"{\"type\":\"io.nats.jetstream.api.v1.consumer_reset_response\",\"stream_name\":\"ORDERS\",\"name\":\"orders-puller\",\"created\":\"2024-01-01T00:00:00Z\",\"config\":{\"deliver_policy\":\"all\",\"ack_policy\":\"explicit\",\"replay_policy\":\"instant\"},\"delivered\":{\"consumer_seq\":1,\"stream_seq\":7},\"ack_floor\":{\"consumer_seq\":0,\"stream_seq\":0},\"num_ack_pending\":0,\"num_redelivered\":0,\"num_waiting\":0,\"num_pending\":3,\"reset_seq\":7}"
consumerListJSON :: LBS.ByteString
consumerListJSON =
"{\"type\":\"io.nats.jetstream.api.v1.consumer_list_response\",\"total\":1,\"offset\":0,\"limit\":1024,\"consumers\":[{\"type\":\"io.nats.jetstream.api.v1.consumer_info_response\",\"stream_name\":\"ORDERS\",\"name\":\"orders-puller\",\"created\":\"2024-01-01T00:00:00Z\",\"config\":{\"deliver_policy\":\"all\",\"ack_policy\":\"explicit\",\"replay_policy\":\"instant\",\"backoff\":[1000000001,2],\"max_batch\":128},\"delivered\":{\"consumer_seq\":2,\"stream_seq\":10},\"ack_floor\":{\"consumer_seq\":1,\"stream_seq\":9},\"num_ack_pending\":1,\"num_redelivered\":0,\"num_waiting\":0,\"num_pending\":3}]}"
consumerNamesJSON :: LBS.ByteString
consumerNamesJSON =
"{\"type\":\"io.nats.jetstream.api.v1.consumer_names_response\",\"total\":2,\"offset\":0,\"limit\":1024,\"consumers\":[\"orders-puller\",\"orders-worker\"]}"
consumerInfoFixture :: ConsumerInfo
consumerInfoFixture = ConsumerInfo
{ consumerInfoStreamName = "ORDERS"
, consumerInfoName = "orders-puller"
, consumerInfoCreated = timestamp
, consumerInfoConfig = ConsumerConfig
{ consumerConfigDurableName = Nothing
, consumerConfigName = Nothing
, consumerConfigDescription = Nothing
, consumerConfigDeliverSubject = Nothing
, consumerConfigDeliverGroup = Nothing
, consumerConfigDeliverPolicy = DeliverAll
, consumerConfigAckPolicy = AckExplicit
, consumerConfigReplayPolicy = ReplayInstant
, consumerConfigFilterSubject = Nothing
, consumerConfigFilterSubjects = Nothing
, consumerConfigAckWait = Nothing
, consumerConfigMaxDeliver = Nothing
, consumerConfigMaxWaiting = Nothing
, consumerConfigMaxAckPending = Nothing
, consumerConfigBackoff = Just [fromRational (1000000001 % 1000000000), fromRational (2 % 1000000000)]
, consumerConfigMaxRequestBatch = Just 128
, consumerConfigInactiveThreshold = Nothing
, consumerConfigIdleHeartbeat = Nothing
, consumerConfigHeadersOnly = Nothing
, consumerConfigReplicas = Nothing
, consumerConfigMemoryStorage = Nothing
}
, consumerInfoDelivered = ConsumerSequenceInfo
{ consumerSequenceConsumer = 2
, consumerSequenceStream = 10
, consumerSequenceLast = Nothing
}
, consumerInfoAckFloor = ConsumerSequenceInfo
{ consumerSequenceConsumer = 1
, consumerSequenceStream = 9
, consumerSequenceLast = Nothing
}
, consumerInfoNumAckPending = 1
, consumerInfoNumRedelivered = 0
, consumerInfoNumWaiting = 0
, consumerInfoNumPending = 3
}
timestamp = read "2024-01-01 00:00:00 UTC"