Merge pull request #256 from BeaudanBrown/p2p-success-respond

Only respond 200 to successful p2p messages

libloki/modules/local_loki_server.js

@@ -57,8 +57,11 @@ class LocalLokiServer extends EventEmitter {
                 sendResponse(STATUS.NOT_FOUND, 'Invalid endpoint!');
                 return;
               }
-              this.emit('message', bodyObject.params.data);
-              sendResponse(STATUS.OK);
+              this.emit('message', {
+                message: bodyObject.params.data,
+                onSuccess: () => sendResponse(STATUS.OK),
+                onFailure: () => sendResponse(STATUS.NOT_FOUND),
+              });
             } catch (e) {
               // Bad Request: Failed to decode json
               sendResponse(STATUS.BAD_REQUEST, 'Failed to decode JSON');

libloki/test/node/local_loki_server_test.js

@@ -64,8 +64,10 @@ describe('LocalLokiServer', () => {
       };
 
       const promise = new Promise(res => {
-        server.on('message', message => {
+        server.on('message', eventData => {
+          const { message, onSuccess } = eventData;
           assert.equal(message, 'This is data');
+          onSuccess();
           server.close();
           res();
         });

libtextsecure/http-resources.js

@@ -77,7 +77,7 @@
       });
     };
 
-    this.handleMessage = (message, isP2p = false) => {
+    this.handleMessage = (message, options = {}) => {
       try {
         const dataPlaintext = stringToArrayBufferBase64(message);
         const messageBuf = textsecure.protobuf.WebSocketMessage.decode(
@@ -93,7 +93,7 @@
               body: messageBuf.request.body,
               id: messageBuf.request.id,
             }),
-            isP2p
+            options
           );
         }
       } catch (error) {

libtextsecure/message_receiver.js

@@ -137,8 +137,13 @@ MessageReceiver.prototype.extend({
       setTimeout(this.startLocalServer.bind(this), 30 * 1000);
     }
   },
-  handleP2pMessage(message) {
-    this.httpPollingResource.handleMessage(message, true);
+  handleP2pMessage({ message, onSuccess, onFailure }) {
+    const options = {
+      isP2p: true,
+      onSuccess,
+      onFailure,
+    };
+    this.httpPollingResource.handleMessage(message, options);
   },
   shutdown() {
     if (this.socket) {
@@ -215,7 +220,8 @@ MessageReceiver.prototype.extend({
     //     return this.dispatchAndWait(event);
     //   });
   },
-  handleRequest(request, isP2p = false) {
+  handleRequest(request, options) {
+    const { isP2p, onSuccess, onFailure } = options;
     this.incoming = this.incoming || [];
     const lastPromise = _.last(this.incoming);
 
@@ -258,7 +264,7 @@ MessageReceiver.prototype.extend({
 
             // To ensure that we queue in the same order we receive messages
             await lastPromise;
-            this.queueEnvelope(envelope);
+            this.queueEnvelope(envelope, onSuccess, onFailure);
           },
           error => {
             request.respond(500, 'Failed to cache message');
@@ -534,7 +540,7 @@ MessageReceiver.prototype.extend({
       );
     });
   },
-  queueEnvelope(envelope) {
+  queueEnvelope(envelope, onSuccess = null, onFailure = null) {
     const id = this.getEnvelopeId(envelope);
     window.log.info('queueing envelope', id);
 
@@ -544,6 +550,11 @@ MessageReceiver.prototype.extend({
       `queueEnvelope ${id}`
     );
     const promise = this.addToQueue(taskWithTimeout);
+    promise.then(() => {
+      if (onSuccess) {
+        onSuccess();
+      }
+    });
 
     return promise.catch(error => {
       window.log.error(
@@ -552,6 +563,9 @@ MessageReceiver.prototype.extend({
         ':',
         error && error.stack ? error.stack : error
       );
+      if (onFailure) {
+        onFailure();
+      }
     });
   },
   // Same as handleEnvelope, just without the decryption step. Necessary for handling