Use separate eventstream per namespace #140
Conversation
Signed-off-by: Nicko Guyer <nicko.guyer@kaleido.io>
Signed-off-by: Nicko Guyer <nicko.guyer@kaleido.io>
Signed-off-by: Nicko Guyer <nicko.guyer@kaleido.io>
Signed-off-by: Nicko Guyer <nicko.guyer@kaleido.io>
Signed-off-by: Nicko Guyer <nicko.guyer@kaleido.io>
Signed-off-by: Nicko Guyer <nicko.guyer@kaleido.io>
Signed-off-by: Nicko Guyer <nicko.guyer@kaleido.io>
Signed-off-by: Nicko Guyer <nicko.guyer@kaleido.io>
| @@ -42,7 +42,7 @@ export class TokensController { | |||
| @HttpCode(204) | |||
| @ApiOperation({ summary: 'Perform one-time initialization (if not auto-initialized)' }) | |||
| init(@RequestContext() ctx: Context) { | |||
| return this.service.init(ctx); | |||
| // Do nothing. Endpoint retained for backwards compatibility with older tooling. | |||
There was a problem hiding this comment.
Will something important (like the FireFly CLI or older FireFly core) crash if we just plain remove this endpoint?
There was a problem hiding this comment.
Yeah I think FireFly CLI calls this, but that call could be removed in new versions
| if (stream !== undefined) { | ||
| return stream; | ||
| } | ||
| private async getStream(ctx: Context, namespace: string) { | ||
| await this.migrationCheck(ctx); |
There was a problem hiding this comment.
Do we still need this migration check? I'm not sure how much value it's adding... might be worth evaluating if it can just be stripped out in this next release.
There was a problem hiding this comment.
Yeah I'm really not sure. I tried to not touch/break as much of the existing "migration" code as possible because I don't fully understand all the iterations it has gone through. But I'm happy to delete more code if it's safe.
There was a problem hiding this comment.
This was there to look for old event streams/subscriptions and give you a nudge in the logs - "hey looks like you may want to delete and clean up old streams".
Since we've now decided to forcefully delete old streams, it doesn't seem useful anymore. It's worth deciding if we need to look for and delete any of the older spellings, in addition to the current stream though.
There was a problem hiding this comment.
I thought some more about this, and I think we should definitely remove it.
I believe it should be part of the release notes - ie if you have an old token connector instance, check its startup logs to be sure there are no warnings logged at startup. If there are warnings logged about any event streams or subscriptions, you should address those warnings before upgrading. The warnings will not be logged after upgrading to this new version.
| this.logger.error(`Error initializing event stream proxy: ${err}`); | ||
| }); | ||
| } | ||
| client.on('message', async (message: string) => { |
There was a problem hiding this comment.
So we're now using a combination of Nest websocket message handlers (ie with @SubscribeMessage('ack')) and raw websocket message handlers (with on('message')). Maybe should choose one or the other?
There was a problem hiding this comment.
Maybe. I'm not sure how nest works 😆
| this.currentClient.send(JSON.stringify(message)); | ||
| } | ||
| } | ||
|
|
||
| @SubscribeMessage('ack') | ||
| handleAck(@MessageBody() data: AckMessageData) { |
There was a problem hiding this comment.
Should we be separating things per websocket client? ie does the client that acks the message need to be the one that we think it was sent to, or is it ok for any connected client to ack any inflight message?
I left a note above about whether we want to use on('message') or SubscribeMessage() in general - if we do opt to keep this message handler, I'll just note you can get the client object with a @ConnectedSocket() client: Socket param.
| this.currentClient = undefined; | ||
| // Iterate over all the namespaces this client was subscribed to | ||
| this.namespaceClients.forEach((clientSet, namespace) => { | ||
| clientSet.delete(client); |
There was a problem hiding this comment.
Shouldn't we be checking the return value to see if this set included this client? Seems like the current behavior will nack all messages on all namespaces, regardless of whether the disconnect client was actually subscribed to them.
Signed-off-by: Nicko Guyer <nicko.guyer@kaleido.io>
Signed-off-by: Nicko Guyer <nicko.guyer@kaleido.io>
Signed-off-by: Nicko Guyer <nicko.guyer@kaleido.io>
| @@ -58,6 +58,10 @@ export interface WebSocketStart extends WebSocketActionBase { | |||
| namespace: string; | |||
There was a problem hiding this comment.
Can probably delete the prior WebSocketMessage type above if we're not using it anywhere.
There was a problem hiding this comment.
This is still used for processing things that come from EVMConnect
| if (!this.awaitingAck.get(client.id)) { | ||
| this.awaitingAck.set(client.id, []); | ||
| } | ||
|
|
||
| client.on('message', async (message: string) => { | ||
| const action = JSON.parse(message) as WebSocketActionBase; | ||
| switch (action.type) { | ||
| case 'start': | ||
| const startAction = action as WebSocketStart; |
There was a problem hiding this comment.
If you create local vars in a case, you may want to wrap with {}
Signed-off-by: Nicko Guyer <nicko.guyer@kaleido.io>
This PR is designed to be compatible with changes in FireFly Core in this PR hyperledger/firefly#1388
Overview of changes:
startcommand for a specific namespace like FireFly Core