-
Notifications
You must be signed in to change notification settings - Fork 1.2k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[EPH] implement CheckpointManager methods (#4583)
* [EPH] implement CheckpointManager methods
- Loading branch information
1 parent
97b684e
commit f1c472c
Showing
9 changed files
with
130 additions
and
89 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,58 +1,93 @@ | ||
| // Copyright (c) Microsoft Corporation. All rights reserved. | ||
| // Licensed under the MIT License. | ||
|
|
||
| // import { PartitionContext } from "./partitionContext"; | ||
| import { EventData } from "./eventData"; | ||
| // import { PartitionManager } from "./eventProcessor"; | ||
| import { PartitionContext } from "./partitionContext"; | ||
| import { ReceivedEventData } from "./eventData"; | ||
| import { PartitionManager } from "./eventProcessor"; | ||
|
|
||
| /** | ||
| * Used by createCheckpoint in PartitionManager | ||
| **/ | ||
| export interface Checkpoint { | ||
| /** | ||
| /** | ||
| * @property The event hub name | ||
| */ | ||
| eventHubName: string; | ||
| /** | ||
| /** | ||
| * @property The consumer group name | ||
| */ | ||
| consumerGroupName: string; | ||
| /** | ||
| /** | ||
| * @property The unique instance identifier | ||
| */ | ||
| instanceId: string; | ||
| /** | ||
| /** | ||
| * @property The identifier of the Event Hub partition | ||
| */ | ||
| partitionId: string; | ||
| /** | ||
| /** | ||
| * @property The sequence number of the event. | ||
| */ | ||
| sequenceNumber: number; | ||
| /** | ||
| /** | ||
| * @property The offset of the event. | ||
| */ | ||
| offset: number; | ||
| /** | ||
| * @property The unique identifier for the operation. | ||
| */ | ||
| eTag: string; | ||
| } | ||
|
|
||
| /** | ||
| * CheckPointManager is created by the library & passed to user's code to let them create a checkpoint | ||
| */ | ||
| export class CheckpointManager { | ||
| // private _partitionContext: PartitionContext; // for internal use by createCheckpoint | ||
| // private _partitionManager: PartitionManager; // for internal use by createCheckpoint | ||
|
|
||
| // constructor(partitionContext: PartitionContext, partitionManager: PartitionManager) { | ||
| // this._partitionContext = partitionContext; | ||
| // this._partitionManager = partitionManager; | ||
| // } | ||
|
|
||
| public async updateCheckpoint(eventData: EventData): Promise<void>; | ||
| private _partitionContext: PartitionContext; | ||
| private _partitionManager: PartitionManager; | ||
| private _instanceId: string; | ||
| private _eTag: string; | ||
|
|
||
| public async updateCheckpoint(offset: string, sequenceNumber: number): Promise<void>; | ||
| constructor(partitionContext: PartitionContext, partitionManager: PartitionManager, instanceId: string) { | ||
| this._partitionContext = partitionContext; | ||
| this._partitionManager = partitionManager; | ||
| this._instanceId = instanceId; | ||
| this._eTag = ""; | ||
| } | ||
| /** | ||
| * Updates a checkpoint using the event data. | ||
| * | ||
| * @param eventData The event data to use for updating the checkpoint. | ||
| * @return Promise<void> | ||
| */ | ||
| public async updateCheckpoint(eventData: ReceivedEventData): Promise<void>; | ||
| /** | ||
| * Updates a checkpoint using the given offset and sequence number. | ||
| * | ||
| * @param sequenceNumber The sequence number to update the checkpoint. | ||
| * @param offset The offset to update the checkpoint. | ||
| * @return Promise<void>. | ||
| */ | ||
| public async updateCheckpoint(sequenceNumber: number, offset: number): Promise<void>; | ||
|
|
||
| public async updateCheckpoint( | ||
| eventDataOrOffset: EventData | string, | ||
| sequenceNumber?: number | ||
| ): Promise<void> {} | ||
| eventDataOrSequenceNumber: ReceivedEventData | number, | ||
| offset?: number | ||
| ): Promise<void> { | ||
| const checkpoint: Checkpoint = { | ||
| eventHubName: this._partitionContext.eventHubName, | ||
| consumerGroupName: this._partitionContext.consumerGroupName, | ||
| instanceId: this._instanceId, | ||
| partitionId: this._partitionContext.partitionId, | ||
| sequenceNumber: | ||
| typeof eventDataOrSequenceNumber === "number" | ||
| ? eventDataOrSequenceNumber | ||
| : eventDataOrSequenceNumber.sequenceNumber, | ||
| offset: | ||
| typeof eventDataOrSequenceNumber === "number" ? offset! : eventDataOrSequenceNumber.offset, | ||
| eTag: this._eTag | ||
| }; | ||
|
|
||
| this._eTag = await this._partitionManager.updateCheckpoint(checkpoint); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.