Skip to content

Direct round trips #68

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
peterbroadhurst merged 1 commit into from
Aug 10, 2022
Merged

Direct round trips #68

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
117 changes: 74 additions & 43 deletions src/routers/api.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,15 +16,17 @@

import { Request, Router } from 'express';
import { promises as fs } from 'fs';
import { createReadStream } from 'fs';
import https from 'https';
import path from 'path';
import { v4 as uuidV4 } from 'uuid';
import * as blobsHandler from '../handlers/blobs';
import * as eventsHandler from '../handlers/events';
import { queueEvent } from '../handlers/events';
import * as messagesHandler from '../handlers/messages';
import { ca, cert, certBundle, key, peerID } from '../lib/cert';
import { config, persistDestinations, persistPeers } from '../lib/config';
import { IStatus } from '../lib/interfaces';
import { IBlobReceivedEvent, IFile, IMessageDeliveredEvent, IMessageReceivedEvent, IStatus } from '../lib/interfaces';
import RequestError from '../lib/request-error';
import * as utils from '../lib/utils';

Expand Down Expand Up @@ -152,46 +154,38 @@ router.post('/messages', async (req, res, next) => {
}
let senderDestination: string | undefined = undefined;
if (typeof req.body.sender === 'string') {
const segments = req.body.sender.split('/');
if (segments[0] !== peerID) {
throw new RequestError(`Sender ID mismatch expected=${peerID} recieved=${segments[0]}`, 400);
let senderID: string;
({ peerID: senderID, destination: senderDestination } = extractRecipientAndDestination(req.body.sender));
if (senderID !== peerID) {
throw new RequestError(`Sender ID mismatch expected=${peerID} recieved=${senderDestination}`, 400);
}
if (segments.length > 1) {
if (!config.destinations?.includes(segments[1])) {
throw new RequestError(`Unknown sender destination expected=${config.destinations?.join('|') ?? 'none'} recieved=${segments[1]}`, 400);
}
senderDestination = segments[1];
if (senderDestination !== undefined && !config.destinations?.includes(senderDestination)) {
throw new RequestError(`Unknown sender destination expected=${config.destinations?.join('|') ?? 'none'} recieved=${senderDestination}`, 400);
Copy link
Contributor

@awrichar awrichar Aug 9, 2022 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

typo: recieved -> received (and a few lines above as well)

}
}
let recipientID: string;
let recipientDestination: string | undefined = undefined;
if (typeof req.body.recipient === 'string') {
const segments = req.body.recipient.split('/');
recipientID = segments[0];
if (segments.length > 1) {
recipientDestination = segments[1];
}
({ peerID: recipientID, destination: recipientDestination } = extractRecipientAndDestination(req.body.recipient));
} else {
throw new RequestError('Missing recipient', 400);
}
let recipientEndpoint: string;
let requestId: string = req.body.requestId ?? uuidV4();
if (recipientID === peerID) {
recipientEndpoint = config.p2p.endpoint ?? `https://${config.p2p.hostname}:${config.p2p.port}`;
if (recipientDestination !== undefined && !config.destinations?.includes(recipientDestination)) {
throw new RequestError(`Unknown recipient destination expected=${config.destinations?.join('|') ?? 'none'} recieved=${recipientDestination}`, 400);
}
dispatchInternalMessage(req.body.sender, req.body.recipient, req.body.message, requestId);
} else {
let recipientPeer = config.peers.find(peer => peer.id === recipientID);
if (recipientPeer === undefined) {
throw new RequestError(`Unknown recipient ${recipientID}`, 400);
}
recipientEndpoint = recipientPeer.endpoint;
if (recipientDestination !== undefined && !recipientPeer.destinations?.includes(recipientDestination)) {
throw new RequestError(`Unknown recipient destination expected=${recipientPeer.destinations?.join('|') ?? 'none'} recieved=${recipientDestination}`, 400);
}
messagesHandler.sendMessage(req.body.message, recipientID, recipientPeer.endpoint, requestId, senderDestination, recipientDestination);
}
let requestId = uuidV4();
if (typeof req.body.requestId === 'string') {
requestId = req.body.requestId;
}
messagesHandler.sendMessage(req.body.message, recipientID, recipientEndpoint, requestId, senderDestination, recipientDestination);
res.send({ requestId });
} catch (err) {
next(err);
Expand Down Expand Up @@ -256,31 +250,25 @@ router.post('/transfers', async (req, res, next) => {
await blobsHandler.retrieveMetadata(req.body.path);
let senderDestination: string | undefined = undefined;
if (typeof req.body.sender === 'string') {
const segments = req.body.sender.split('/');
if (segments[0] !== peerID) {
throw new RequestError(`Sender ID mismatch expected=${peerID} recieved=${segments[0]}`, 400);
let senderID: string;
({ peerID: senderID, destination: senderDestination } = extractRecipientAndDestination(req.body.sender));
if (senderID !== peerID) {
throw new RequestError(`Sender ID mismatch expected=${peerID} recieved=${senderID}`, 400);
}
if (segments.length > 1) {
if (!config.destinations?.includes(segments[1])) {
throw new RequestError(`Unknown sender destination expected=${config.destinations?.join('|')} recieved=${segments[1]}`, 400);
}
senderDestination = segments[1];
if (senderDestination !== undefined && !config.destinations?.includes(senderDestination)) {
throw new RequestError(`Unknown sender destination expected=${config.destinations?.join('|')} recieved=${senderDestination}`, 400);
}
}
let recipientID: string;
let recipientDestination: string | undefined = undefined;
if (typeof req.body.recipient === 'string') {
const segments = req.body.recipient.split('/');
recipientID = segments[0];
if (segments.length > 1) {
recipientDestination = segments[1];
}
({ peerID: recipientID, destination: recipientDestination } = extractRecipientAndDestination(req.body.recipient));
} else {
throw new RequestError('Missing recipient', 400);
}
let recipientEndpoint: string;
let requestId: string = req.body.requestId ?? uuidV4();
if (recipientID === peerID) {
recipientEndpoint = config.p2p.endpoint ?? `https://${config.p2p.hostname}:${config.p2p.port}`;
dispatchInternalBlob(req.body.sender, req.body.recipient, req.body.path);
} else {
let recipientPeer = config.peers.find(peer => peer.id === recipientID);
if (recipientPeer === undefined) {
Expand All @@ -289,15 +277,58 @@ router.post('/transfers', async (req, res, next) => {
if (recipientDestination !== undefined && !recipientPeer.destinations?.includes(recipientDestination)) {
throw new RequestError(`Unknown recipient destination expected=${recipientPeer.destinations?.join('|')} recieved=${recipientDestination}`, 400);
}
recipientEndpoint = recipientPeer.endpoint;
}
let requestId = uuidV4();
if (typeof req.body.requestId === 'string') {
requestId = req.body.requestId;
blobsHandler.sendBlob(req.body.path, recipientID, recipientPeer.endpoint, requestId, senderDestination, recipientDestination);
}
blobsHandler.sendBlob(req.body.path, recipientID, recipientEndpoint, requestId, senderDestination, recipientDestination);
res.send({ requestId });
} catch (err) {
next(err);
}
});

const extractRecipientAndDestination = (value: string) => {
const segments = value.split('/');
return {
peerID: segments[0],
destination: segments.length > 1 ? segments[1] : undefined
};
};

const dispatchInternalMessage = async (sender: string, recipient: string, message: string, requestId: string) => {
await queueEvent({
id: uuidV4(),
type: 'message-received',
sender,
recipient,
message
} as IMessageReceivedEvent);
await queueEvent({
id: uuidV4(),
type: 'message-delivered',
sender,
recipient,
message,
requestId
} as IMessageDeliveredEvent);
};

const dispatchInternalBlob = async (sender: string, recipient: string, filePath: string) => {
const originBlobPath = path.join(utils.constants.DATA_DIRECTORY, utils.constants.BLOBS_SUBDIRECTORY, filePath);
const readableStream = createReadStream(originBlobPath);
const file: IFile = {
key: '',
name: '',
readableStream
};
const destinationBlobPath = path.join(utils.constants.RECEIVED_BLOBS_SUBDIRECTORY, sender, filePath);
const metadata = await blobsHandler.storeBlob(file, destinationBlobPath);
await queueEvent({
id: uuidV4(),
type: 'blob-received',
sender,
recipient,
path: destinationBlobPath,
hash: metadata.hash,
size: metadata.size,
lastUpdate: metadata.lastUpdate
} as IBlobReceivedEvent);
};