fix: after receive message pipe

and add check to discover similar issues in the future

2 files changed, 26 insertions, 14 deletions

diff --git a/src/plugins/NotificationPlugin.ts b/src/plugins/NotificationPlugin.ts
index 471fc1b7..1ba6cccc 100644
--- a/src/plugins/NotificationPlugin.ts
+++ b/src/plugins/NotificationPlugin.ts
@@ -1,11 +1,11 @@
-import Message from '../Message';
 import { AbstractPlugin } from '../plugin/AbstractPlugin';
 import PluginAPI from '../plugin/PluginAPI';
-import Contact from '../Contact';
 import Translation from '../util/Translation';
 import Notification from '../Notification';
 import { Presence } from '../connection/AbstractConnection';
 import { SOUNDS } from '../CONST';
+import { IContact } from '@src/Contact.interface';
+import { IMessage } from '@src/Message.interface';
 
 const MIN_VERSION = '4.0.0';
 const MAX_VERSION = '99.0.0';
@@ -32,7 +32,11 @@ export default class NotificationPlugin extends AbstractPlugin {
       pluginAPI.registerPresenceHook(this.onPresence);
    }
 
-   private afterReceiveMessageProcessor = (contact: Contact, message: Message): Promise<any> => {
+   private afterReceiveMessageProcessor = (
+      contact: IContact,
+      message: IMessage,
+      stanza: Element
+   ): Promise<[IContact, IMessage, Element]> => {
       if ((message.getPlaintextMessage() || message.getAttachment()) && message.isIncoming()) {
          Notification.notify({
             title: Translation.t('New_message_from', {
@@ -44,10 +48,10 @@ export default class NotificationPlugin extends AbstractPlugin {
          });
       }
 
-      return Promise.resolve([contact, message]);
+      return Promise.resolve([contact, message, stanza]);
    };
 
-   private onPresence = (contact: Contact, newPresence, oldPresence) => {
+   private onPresence = (contact: IContact, newPresence, oldPresence) => {
       if (oldPresence !== Presence.offline || newPresence === Presence.offline) {
          return;
       }
diff --git a/src/util/Pipe.ts b/src/util/Pipe.ts
index 359bba6a..95dad395 100644
--- a/src/util/Pipe.ts
+++ b/src/util/Pipe.ts
@@ -1,14 +1,14 @@
+import Log from './Log';
+
 const MAX_PRIORITY = 100;
 const MIN_PRIORITY = 0;
 
-type Params = any[];
-
-export default class Pipe<params extends Params = any[]> {
-   private pipe = [];
+export default class Pipe<Params extends any[] = any[]> {
+   private pipe: ((...args: Params) => Promise<Params> | Params)[][] = [];
 
    constructor() {}
 
-   public addProcessor(processor: (...args: params) => Promise<params> | params, priority: number = 50) {
+   public addProcessor(processor: (...args: Params) => Promise<Params> | Params, priority: number = 50) {
       if (isNaN(priority) || priority < MIN_PRIORITY || priority > MAX_PRIORITY) {
          throw new Error('Priority has to be between 0 and 100');
       }
@@ -20,7 +20,7 @@ export default class Pipe<params extends Params = any[]> {
       this.pipe[priority].push(processor);
    }
 
-   public run(...args: params): Promise<params> {
+   public run(...args: Params): Promise<Params> {
       let chain = Promise.resolve(args);
 
       this.pipe.forEach(processors => {
@@ -34,9 +34,17 @@ export default class Pipe<params extends Params = any[]> {
          }
 
          processors.forEach(processor => {
-            chain = chain.then((args2: any[]) => {
-               return processor.apply(this, args2);
-            });
+            chain = chain
+               .then((args2: Params) => {
+                  return processor.apply(this, args2);
+               })
+               .then(args3 => {
+                  if (args.length !== args3.length) {
+                     Log.warn('Bad processor detected', processor);
+                  }
+
+                  return args3;
+               });
          });
       });