@@ -29,14 +29,14 @@ import ko from 'knockout';
29
29
import { container } from 'tsyringe' ;
30
30
31
31
import { Account , ConnectionState , ProcessedEventPayload } from '@wireapp/core' ;
32
+ import { PromiseQueue } from '@wireapp/promise-queue' ;
32
33
import { WebAppEvents } from '@wireapp/webapp-events' ;
33
34
34
35
import { ClientConversationEvent , EventBuilder } from 'Repositories/conversation/EventBuilder' ;
35
36
import { CryptographyMapper } from 'Repositories/cryptography/CryptographyMapper' ;
36
37
import { EventName } from 'Repositories/tracking/EventName' ;
37
38
import { UserState } from 'Repositories/user/UserState' ;
38
39
import { getLogger , Logger } from 'Util/Logger' ;
39
- import { queue } from 'Util/PromiseQueue' ;
40
40
import { TIME_IN_MILLIS } from 'Util/TimeUtil' ;
41
41
42
42
import { ClientEvent } from './Client' ;
@@ -63,6 +63,8 @@ export class EventRepository {
63
63
/** event processors are classes that are able to react and process an incoming event */
64
64
private eventProcessors : EventProcessor [ ] = [ ] ;
65
65
66
+ private eventQueue : PromiseQueue = new PromiseQueue ( ) ;
67
+
66
68
static get CONFIG ( ) {
67
69
return {
68
70
E_CALL_EVENT_LIFETIME : TIME_IN_MILLIS . SECOND * 30 ,
@@ -160,17 +162,19 @@ export class EventRepository {
160
162
* this function will process any incoming event. It is being queued in case 2 events arrive at the same time.
161
163
* Processing events should happen sequentially (thus the queue)
162
164
*/
163
- private readonly handleIncomingEvent = queue ( async ( payload : HandledEventPayload , source : NotificationSource ) => {
164
- try {
165
- await this . handleEvent ( payload , source ) ;
166
- } catch ( error ) {
167
- if ( source === EventSource . NOTIFICATION_STREAM ) {
168
- this . logger . warn ( `Failed to handle event of type "${ event . type } ": ${ error . message } ` , error ) ;
169
- } else {
170
- throw error ;
165
+ private readonly handleIncomingEvent = async ( payload : HandledEventPayload , source : NotificationSource ) => {
166
+ return this . eventQueue . push ( async ( ) => {
167
+ try {
168
+ await this . handleEvent ( payload , source ) ;
169
+ } catch ( error ) {
170
+ if ( source === EventSource . NOTIFICATION_STREAM ) {
171
+ this . logger . warn ( `Failed to handle event of type "${ event . type } ": ${ error . message } ` , error ) ;
172
+ } else {
173
+ throw error ;
174
+ }
171
175
}
172
- }
173
- } ) ;
176
+ } ) ;
177
+ } ;
174
178
175
179
/**
176
180
* Import events coming from an external source. This is only useful useful for profiling or debugging.
@@ -324,7 +328,7 @@ export class EventRepository {
324
328
}
325
329
326
330
const conversationId = 'conversation' in event && event . conversation ;
327
- const inSelfConversation = conversationId === this . userState . self ( ) . id ;
331
+ const inSelfConversation = conversationId === this . userState . self ( ) ? .id ;
328
332
if ( ! inSelfConversation ) {
329
333
return this . processEvent ( event , source ) ;
330
334
}
0 commit comments