fix: add config to process messages concurrently #2026
Conversation
|
sairanjit
force-pushed
the
feat/concurrent-message-processing
branch
from
da1b2a1
to
3281bc5
Compare
| filter((e) => e.payload.message.id === encryptedMessage.id), | ||
| filter((e) => e.payload.message.type === encryptedMessage.type), |
There was a problem hiding this comment.
an encrypted message does not have an id and type right?
There was a problem hiding this comment.
Sorry, I missed this
There was a problem hiding this comment.
- But here there are no common properties I can map to. Can you suggest me a way to add a guard here or filter specific encrypted message?
…ncurrently Signed-off-by: Pritam Singh <[email protected]>
Signed-off-by: Sai Ranjit Tummalapalli <[email protected]>
Signed-off-by: Sai Ranjit Tummalapalli <[email protected]>
Signed-off-by: Sai Ranjit Tummalapalli <[email protected]>
Signed-off-by: Sai Ranjit Tummalapalli <[email protected]>
sairanjit
force-pushed
the
feat/concurrent-message-processing
branch
from
05caec4
to
9a37419
Compare
Signed-off-by: Sai Ranjit Tummalapalli <[email protected]>
Signed-off-by: Sai Ranjit Tummalapalli <[email protected]>
| observable | ||
| .pipe( | ||
| filter((e) => e.type === AgentEventTypes.AgentMessageProcessed), | ||
| filter((e) => deepEquality(e.payload.encryptedMessage, encryptedMessage)), |
There was a problem hiding this comment.
I think we can just check for equality right (===)? AS the object will be passed around? Using deep object equality can be expensive
| @@ -64,7 +65,7 @@ class Dispatcher { | |||
| await next() | |||
| } | |||
|
|
|||
| public async dispatch(messageContext: InboundMessageContext): Promise<void> { | |||
| public async dispatch(messageContext: InboundMessageContext, encryptedMessage?: EncryptedMessage): Promise<void> { | |||
There was a problem hiding this comment.
What if we add encryptedMessage to the InboundMessageContext ( as it's related to the incoming message)
| filter((e) => e.type === AgentEventTypes.AgentMessageProcessed), | ||
| filter((e) => deepEquality(e.payload.encryptedMessage, encryptedMessage)), | ||
| timeout({ | ||
| first: 10000, // timeout after 10 seconds |
There was a problem hiding this comment.
I think it's less relevant here to await the processing? As we don't do anything afterwards
| filter((e) => e.type === AgentEventTypes.AgentMessageProcessed), | ||
| filter((e) => deepEquality(e.payload.encryptedMessage, encryptedMessage)), | ||
| timeout({ | ||
| first: 10000, // timeout after 10 seconds |
There was a problem hiding this comment.
Let's add this as a param to the constructor of the HttpInboundTransport class?
Signed-off-by: Sai Ranjit Tummalapalli <[email protected]>
|
@sairanjit can you resolve the conflicts? I think we're good to merge! |
…t/concurrent-message-processing Signed-off-by: Sai Ranjit Tummalapalli <[email protected]>
|
@TimoGlastra Resolved conflicts now |
…on#2026) Signed-off-by: Pritam Singh <[email protected]> Signed-off-by: Sai Ranjit Tummalapalli <[email protected]> Co-authored-by: Pritam Singh <[email protected]> Signed-off-by: Martin Auer <[email protected]>
cc: @Zzocker : Taking it further from #1276
@TimoGlastra From the earlier conversations how can we handle the queue from inbound transport?