diff --git a/nakadi-client.cabal b/nakadi-client.cabal --- a/nakadi-client.cabal +++ b/nakadi-client.cabal @@ -2,10 +2,10 @@ -- -- see: https://github.com/sol/hpack -- --- hash: 61b5d898775025659f82db56b3249871d299f52224409c2b8eb75d1d4adc90ca +-- hash: 74ff60305e7890a09afb58a9d855f782f960dab69c2ca9901570da82116a91f2 name: nakadi-client -version: 0.6.0.0 +version: 0.6.1.0 synopsis: Client library for the Nakadi Event Broker description: This package implements a client library for interacting with the Nakadi event broker system developed by Zalando. category: Network diff --git a/src/Network/Nakadi/Internal/Committer/TimeBuffer.hs b/src/Network/Nakadi/Internal/Committer/TimeBuffer.hs --- a/src/Network/Nakadi/Internal/Committer/TimeBuffer.hs +++ b/src/Network/Nakadi/Internal/Committer/TimeBuffer.hs @@ -46,8 +46,8 @@ Timer.withAsyncTimer timerConf $ \timer -> forever $ do Timer.wait timer commitAllCursors identity eventStream cursorsMap - where -- | The cursorsConsumer drains the cursors queue and adds each + where -- The cursorsConsumer drains the cursors queue and adds each -- cursor to the provided cursorsMap. - cursorConsumer cursorsMap = forever . liftIO . atomically $ do - (_, cursor) <- readTBQueue queue - modifyTVar cursorsMap (HashMap.insert (cursorKey cursor) cursor) + cursorConsumer cursorsMap = forever . liftIO . atomically $ do + (_, cursor) <- readTBQueue queue + modifyTVar cursorsMap (HashMap.insert (cursorKey cursor) cursor) diff --git a/src/Network/Nakadi/Internal/Types/Service.hs b/src/Network/Nakadi/Internal/Types/Service.hs --- a/src/Network/Nakadi/Internal/Types/Service.hs +++ b/src/Network/Nakadi/Internal/Types/Service.hs @@ -85,6 +85,21 @@ parseJSON (String name) = return $ PartitionName name parseJSON invalid = typeMismatch "PartitionName" invalid +-- | Type for partition compaction keys. +newtype PartitionCompactionKey = PartitionCompactionKey + { unPartitionCompactionKey :: Text -- ^ Wrapped Partition Compaction Key + } deriving (Eq, Show, Generic, Ord, Hashable) + +instance IsString PartitionCompactionKey where + fromString = PartitionCompactionKey . Text.pack + +instance ToJSON PartitionCompactionKey where + toJSON = String . unPartitionCompactionKey + +instance FromJSON PartitionCompactionKey where + parseJSON (String name) = return $ PartitionCompactionKey name + parseJSON invalid = typeMismatch "PartitionCompactionKey" invalid + -- | Type for cursor tokens. newtype CursorToken = CursorToken Text deriving (Eq, Show, Ord) @@ -768,6 +783,24 @@ "none" -> return CompatibilityModeNone invalid -> typeMismatch "CompatibilityMode" invalid + +-- | Type for cleanup policy. + +data CleanupPolicy = CleanupPolicyDelete + | CleanupPolicyCompact + deriving (Show, Eq, Ord, Generic, Hashable) + +instance ToJSON CleanupPolicy where + toJSON policy = String $ case policy of + CleanupPolicyDelete -> "delete" + CleanupPolicyCompact -> "compact" + +instance FromJSON CleanupPolicy where + parseJSON strategy = case strategy of + "delete" -> return CleanupPolicyDelete + "compact" -> return CleanupPolicyCompact + invalid -> typeMismatch "CleanupPolicy" invalid + -- | Type for a partitioning key field. newtype PartitionKeyField = PartitionKeyField { unPartitionKeyField :: Text } @@ -813,6 +846,7 @@ , _compatibilityMode :: Maybe CompatibilityMode , _schema :: EventTypeSchema , _partitionKeyFields :: Maybe [PartitionKeyField] + , _cleanupPolicy :: Maybe CleanupPolicy , _defaultStatistic :: Maybe EventTypeStatistics , _options :: Maybe EventTypeOptions } deriving (Show, Generic, Eq, Ord, Hashable) @@ -822,10 +856,11 @@ -- | Type of published event metadata values. data EventMetadata = EventMetadata - { _eid :: EventId - , _occurredAt :: Timestamp - , _parentEids :: Maybe [EventId] - , _partition :: Maybe PartitionName + { _eid :: EventId + , _occurredAt :: Timestamp + , _parentEids :: Maybe [EventId] + , _partition :: Maybe PartitionName + , _partitionCompactionKey :: Maybe PartitionCompactionKey } deriving (Eq, Show, Generic) deriveJSON nakadiJsonOptions ''EventMetadata @@ -833,14 +868,15 @@ -- | Type of event metadata enriched by Nakadi data EventMetadataEnriched = EventMetadataEnriched - { _eid :: EventId - , _eventType :: EventTypeName - , _occurredAt :: Timestamp - , _receivedAt :: Timestamp - , _version :: SchemaVersion - , _parentEids :: Maybe [EventId] - , _flowId :: Maybe FlowId - , _partition :: Maybe PartitionName + { _eid :: EventId + , _eventType :: EventTypeName + , _occurredAt :: Timestamp + , _receivedAt :: Timestamp + , _version :: SchemaVersion + , _parentEids :: Maybe [EventId] + , _flowId :: Maybe FlowId + , _partition :: Maybe PartitionName + , _partitionCompactionKey :: Maybe PartitionCompactionKey } deriving (Eq, Show, Generic) deriveJSON nakadiJsonOptions ''EventMetadataEnriched diff --git a/src/Network/Nakadi/Types/Service.hs b/src/Network/Nakadi/Types/Service.hs --- a/src/Network/Nakadi/Types/Service.hs +++ b/src/Network/Nakadi/Types/Service.hs @@ -14,6 +14,7 @@ ( CursorOffset(..) , EventTypeName(..) , PartitionName(..) + , PartitionCompactionKey(..) , CursorToken(..) , Cursor(..) , ApplicationName(..) @@ -61,6 +62,7 @@ , PartitionStrategy(..) , EnrichmentStrategy(..) , CompatibilityMode(..) + , CleanupPolicy(..) , PartitionKeyField(..) , EventType(..) , DataChangeEvent(..) diff --git a/tests/Network/Nakadi/EventTypes/BusinessEvents/Test.hs b/tests/Network/Nakadi/EventTypes/BusinessEvents/Test.hs --- a/tests/Network/Nakadi/EventTypes/BusinessEvents/Test.hs +++ b/tests/Network/Nakadi/EventTypes/BusinessEvents/Test.hs @@ -41,6 +41,7 @@ , _enrichmentStrategies = Just [EnrichmentStrategyMetadata] , _partitionStrategy = Nothing , _compatibilityMode = Nothing + , _cleanupPolicy = Nothing , _schema = EventTypeSchema { _version = Nothing , _createdAt = Nothing @@ -70,10 +71,11 @@ pure BusinessEvent { _payload = payload , _metadata = EventMetadata - { _eid = eid - , _occurredAt = timestamp - , _parentEids = Nothing - , _partition = Nothing + { _eid = eid + , _occurredAt = timestamp + , _parentEids = Nothing + , _partition = Nothing + , _partitionCompactionKey = Nothing } } diff --git a/tests/Network/Nakadi/EventTypes/Test.hs b/tests/Network/Nakadi/EventTypes/Test.hs --- a/tests/Network/Nakadi/EventTypes/Test.hs +++ b/tests/Network/Nakadi/EventTypes/Test.hs @@ -123,10 +123,11 @@ let event = DataChangeEvent { _payload = Foo "Hello!" , _metadata = EventMetadata - { _eid = eid - , _occurredAt = Timestamp now - , _parentEids = Nothing - , _partition = Nothing + { _eid = eid + , _occurredAt = Timestamp now + , _parentEids = Nothing + , _partition = Nothing + , _partitionCompactionKey = Nothing } , _dataType = "test.FOO" , _dataOp = DataOpUpdate @@ -159,10 +160,11 @@ let event = DataChangeEvent { _payload = Foo "Hello!" , _metadata = EventMetadata - { _eid = eid - , _occurredAt = Timestamp now - , _parentEids = Nothing - , _partition = Nothing + { _eid = eid + , _occurredAt = Timestamp now + , _parentEids = Nothing + , _partition = Nothing + , _partitionCompactionKey = Nothing } , _dataType = "test.FOO" , _dataOp = DataOpUpdate @@ -186,10 +188,11 @@ let event = DataChangeEvent { _payload = Foo "Hello!" , _metadata = EventMetadata - { _eid = eid - , _occurredAt = Timestamp now - , _parentEids = Nothing - , _partition = Nothing + { _eid = eid + , _occurredAt = Timestamp now + , _parentEids = Nothing + , _partition = Nothing + , _partitionCompactionKey = Nothing } , _dataType = "test.FOO" , _dataOp = DataOpUpdate @@ -215,10 +218,11 @@ let event = DataChangeEvent { _payload = Foo "Hello!" , _metadata = EventMetadata - { _eid = eid - , _occurredAt = Timestamp now - , _parentEids = Nothing - , _partition = Nothing + { _eid = eid + , _occurredAt = Timestamp now + , _parentEids = Nothing + , _partition = Nothing + , _partitionCompactionKey = Nothing } , _dataType = "test.FOO" , _dataOp = DataOpUpdate diff --git a/tests/Network/Nakadi/Examples/Subscription/Test.hs b/tests/Network/Nakadi/Examples/Subscription/Test.hs --- a/tests/Network/Nakadi/Examples/Subscription/Test.hs +++ b/tests/Network/Nakadi/Examples/Subscription/Test.hs @@ -24,10 +24,11 @@ let event = Nakadi.DataChangeEvent { Nakadi._payload = Foo "Hello!" , Nakadi._metadata = Nakadi.EventMetadata - { Nakadi._eid = eid - , Nakadi._occurredAt = Nakadi.Timestamp now - , Nakadi._parentEids = Nothing - , Nakadi._partition = Nothing + { Nakadi._eid = eid + , Nakadi._occurredAt = Nakadi.Timestamp now + , Nakadi._parentEids = Nothing + , Nakadi._partition = Nothing + , Nakadi._partitionCompactionKey = Nothing } , Nakadi._dataType = "test.FOO" , Nakadi._dataOp = Nakadi.DataOpUpdate diff --git a/tests/Network/Nakadi/Tests/Common.hs b/tests/Network/Nakadi/Tests/Common.hs --- a/tests/Network/Nakadi/Tests/Common.hs +++ b/tests/Network/Nakadi/Tests/Common.hs @@ -61,6 +61,7 @@ , _partitionStrategy = Just "hash" , _compatibilityMode = Just CompatibilityModeForward , _partitionKeyFields = Just ["fortune"] + , _cleanupPolicy = Just CleanupPolicyDelete , _schema = myEventTypeSchema , _defaultStatistic = Just EventTypeStatistics { _messagesPerMinute = 1000 @@ -82,10 +83,11 @@ myDataChangeEvent eid now = DataChangeEvent { _payload = Foo "Hello!" , _metadata = EventMetadata - { _eid = eid - , _occurredAt = Timestamp now - , _parentEids = Nothing - , _partition = Nothing + { _eid = eid + , _occurredAt = Timestamp now + , _parentEids = Nothing + , _partition = Nothing + , _partitionCompactionKey = Nothing } , _dataType = "test.FOO" , _dataOp = DataOpUpdate @@ -99,10 +101,11 @@ pure DataChangeEvent { _payload = Foo "Hello!" , _metadata = EventMetadata - { _eid = EventId eid - , _occurredAt = Timestamp now - , _parentEids = Nothing - , _partition = Nothing + { _eid = EventId eid + , _occurredAt = Timestamp now + , _parentEids = Nothing + , _partition = Nothing + , _partitionCompactionKey = Nothing } , _dataType = "test.FOO" , _dataOp = DataOpUpdate @@ -115,10 +118,11 @@ pure DataChangeEvent { _payload = Foo ("Hello " ++ Text.pack (show idx)) , _metadata = EventMetadata - { _eid = EventId eid - , _occurredAt = Timestamp now - , _parentEids = Nothing - , _partition = Nothing + { _eid = EventId eid + , _occurredAt = Timestamp now + , _parentEids = Nothing + , _partition = Nothing + , _partitionCompactionKey = Nothing } , _dataType = "test.FOO" , _dataOp = DataOpUpdate