Block waiting for dispatch so that HTTP calls push back to sender on queue full

Signed-off-by: Peter Broadhurst <[email protected]>

Author: peterbroadhurst

src/app.ts

@@ -21,17 +21,15 @@ import path from 'path';
 import swaggerUi from 'swagger-ui-express';
 import WebSocket from 'ws';
 import YAML from 'yamljs';
-import { eventEmitter as blobsEventEmitter } from './handlers/blobs';
 import * as eventsHandler from './handlers/events';
-import { eventEmitter as messagesEventEmitter } from './handlers/messages';
 import { genTLSContext, init as initCert, loadPeerCAs } from './lib/cert';
 import { config, init as initConfig } from './lib/config';
 import { IAckEvent } from './lib/interfaces';
 import { Logger } from './lib/logger';
 import RequestError, { errorHandler } from './lib/request-error';
 import * as utils from './lib/utils';
 import { router as apiRouter, setRefreshCACerts } from './routers/api';
-import { eventEmitter as p2pEventEmitter, router as p2pRouter } from './routers/p2p';
+import { router as p2pRouter } from './routers/p2p';
 import { init as initEvents } from './handlers/events';
 
 const log = new Logger("app.ts");
@@ -71,10 +69,6 @@ export const start = async () => {
     }
   });
 
-  p2pEventEmitter.addListener('event', event => eventsHandler.queueEvent(event));
-  blobsEventEmitter.addListener('event', event => eventsHandler.queueEvent(event));
-  messagesEventEmitter.addListener('event', event => eventsHandler.queueEvent(event));
-
   eventsHandler.getEmitter().addListener('event', event => {
     log.info(`Event emitted ${event.type}/${event.id}`)
     if (delegatedWebSocket !== undefined) {

src/handlers/blobs.ts

@@ -15,7 +15,6 @@
 // limitations under the License.
 
 import crypto from 'crypto';
-import EventEmitter from 'events';
 import FormData from 'form-data';
 import { createReadStream, createWriteStream, promises as fs } from 'fs';
 import https from 'https';
@@ -27,12 +26,12 @@ import { BlobTask, IBlobDeliveredEvent, IBlobFailedEvent, IFile, IMetadata } fro
 import { Logger } from '../lib/logger';
 import RequestError from '../lib/request-error';
 import * as utils from '../lib/utils';
+import { queueEvent } from './events';
 
 const log = new Logger("handlers/blobs.ts")
 
 let blobQueue: BlobTask[] = [];
 let sending = false;
-export const eventEmitter = new EventEmitter();
 
 export const retreiveBlob = async (filePath: string) => {
   const resolvedFilePath = path.join(utils.constants.DATA_DIRECTORY, utils.constants.BLOBS_SUBDIRECTORY, filePath);
@@ -100,7 +99,7 @@ export const deliverBlob = async ({ blobPath, recipient, recipientURL, requestId
       timeout: utils.constants.REST_API_CALL_BLOB_REQUEST_TIMEOUT,
       httpsAgent
     });
-    eventEmitter.emit('event', {
+    await queueEvent({
       id: uuidV4(),
       type: 'blob-delivered',
       path: blobPath,
@@ -109,7 +108,7 @@ export const deliverBlob = async ({ blobPath, recipient, recipientURL, requestId
     } as IBlobDeliveredEvent);
     log.trace(`Blob delivered`);
   } catch (err: any) {
-    eventEmitter.emit('event', {
+    await queueEvent({
       id: uuidV4(),
       type: 'blob-failed',
       path: blobPath,

src/handlers/messages.ts

@@ -14,20 +14,19 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-import EventEmitter from 'events';
 import FormData from 'form-data';
 import https from 'https';
 import { v4 as uuidV4 } from 'uuid';
 import { ca, cert, key } from '../lib/cert';
 import { IMessageDeliveredEvent, IMessageFailedEvent, MessageTask } from '../lib/interfaces';
 import { Logger } from '../lib/logger';
 import * as utils from '../lib/utils';
+import { queueEvent } from './events';
 
 const log = new Logger('handlers/messages.ts')
 
 let messageQueue: MessageTask[] = [];
 let sending = false;
-export const eventEmitter = new EventEmitter();
 
 export const sendMessage = async (message: string, recipient: string, recipientURL: string, requestId: string | undefined) => {
   if (sending) {
@@ -55,7 +54,7 @@ export const deliverMessage = async ({ message, recipient, recipientURL, request
       headers: formData.getHeaders(),
       httpsAgent
     });
-    eventEmitter.emit('event', {
+    await queueEvent({
       id: uuidV4(),
       type: 'message-delivered',
       message,
@@ -64,7 +63,7 @@ export const deliverMessage = async ({ message, recipient, recipientURL, request
     } as IMessageDeliveredEvent);
     log.trace(`Message delivered`);
   } catch(err: any) {
-    eventEmitter.emit('event', {
+    await queueEvent({
       id: uuidV4(),
       type: 'message-failed',
       message,

src/routers/p2p.ts

@@ -18,12 +18,11 @@ import { Router, Request } from 'express';
 import * as utils from '../lib/utils';
 import * as blobsHandler from '../handlers/blobs';
 import path from 'path';
-import { EventEmitter } from 'events';
 import { IBlobReceivedEvent, IMessageReceivedEvent } from '../lib/interfaces';
 import { v4 as uuidV4 } from 'uuid';
+import { queueEvent } from '../handlers/events';
 
 export const router = Router();
-export const eventEmitter = new EventEmitter();
 
 router.head('/ping', (_req, res) => {
   res.sendStatus(204);
@@ -33,7 +32,7 @@ router.post('/messages', async (req: Request, res, next) => {
   try {
     const sender = utils.extractPeerSenderFromRequest(req);
     const message = await utils.extractMessageFromMultipartForm(req);
-    eventEmitter.emit('event', {
+    await queueEvent({
       id: uuidV4(),
       type: 'message-received',
       sender,
@@ -52,7 +51,7 @@ router.put('/blobs/*', async (req: Request, res, next) => {
     const blobPath = path.join(utils.constants.RECEIVED_BLOBS_SUBDIRECTORY, sender, req.params[0]);
     const metadata = await blobsHandler.storeBlob(file, blobPath);
     res.sendStatus(204);
-    eventEmitter.emit('event', {
+    await queueEvent({
       id: uuidV4(),
       type: 'blob-received',
       sender,