nakadi-client 0.6.0.0 → 0.6.1.0
raw patch · 8 files changed
+103/−54 lines, 8 files
Files
- nakadi-client.cabal +2/−2
- src/Network/Nakadi/Internal/Committer/TimeBuffer.hs +4/−4
- src/Network/Nakadi/Internal/Types/Service.hs +48/−12
- src/Network/Nakadi/Types/Service.hs +2/−0
- tests/Network/Nakadi/EventTypes/BusinessEvents/Test.hs +6/−4
- tests/Network/Nakadi/EventTypes/Test.hs +20/−16
- tests/Network/Nakadi/Examples/Subscription/Test.hs +5/−4
- tests/Network/Nakadi/Tests/Common.hs +16/−12
nakadi-client.cabal view
@@ -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
src/Network/Nakadi/Internal/Committer/TimeBuffer.hs view
@@ -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)
src/Network/Nakadi/Internal/Types/Service.hs view
@@ -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
src/Network/Nakadi/Types/Service.hs view
@@ -14,6 +14,7 @@ ( CursorOffset(..) , EventTypeName(..) , PartitionName(..)+ , PartitionCompactionKey(..) , CursorToken(..) , Cursor(..) , ApplicationName(..)@@ -61,6 +62,7 @@ , PartitionStrategy(..) , EnrichmentStrategy(..) , CompatibilityMode(..)+ , CleanupPolicy(..) , PartitionKeyField(..) , EventType(..) , DataChangeEvent(..)
tests/Network/Nakadi/EventTypes/BusinessEvents/Test.hs view
@@ -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 } }
tests/Network/Nakadi/EventTypes/Test.hs view
@@ -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
tests/Network/Nakadi/Examples/Subscription/Test.hs view
@@ -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
tests/Network/Nakadi/Tests/Common.hs view
@@ -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