diff --git a/Readme.md b/Readme.md
index e98758e054d7fbf6ed340d82cdce670b2814f3a6..bb16672ed9a36dfe566251ac3e9b12e6b0c65412 100644
--- a/Readme.md
+++ b/Readme.md
@@ -46,6 +46,7 @@ Configuration is done via Environment variables. In case the variable is not set
|Variable|Description|Default Value|Required|
|---|---|---|---|
+|MODE|Specifies the services that the instance will start. Can be `development`, `ingress`, `ingestion` or `api`. `development` mode will start all services, at the cost of performance. **Only one `ingress` service must be active at any time for each MQTT Broker. Otherwise duplicate data might be put into the database.** `ingestion` mode workers can be scaled up as needed. All modes require access to the database, MQTT and RabbitMQ.||*|
|API_PORT|The port the API will listen on|`8080`||
|API_BASE_URL|The base URL that the API will be accessible on. Example: `https://example.com`||*|
|MQTT_HOST|The host the MQTT Client will connect to|`localhost`||
@@ -65,7 +66,6 @@ Configuration is done via Environment variables. In case the variable is not set
|RABBITMQ_PASSWORD|The password the RabbitMQ Client will use|||
|RABBITMQ_QUEUE|The name of the queue the RabbitMQ Client will use. This will be created if it does not exist.|`measurements`||
-RABBITMQ_QUEUE
## API Documentation
Some of the mentioned endpoints support versioning by setting a `Accept-Datetime` header according to the HTTP Memento specification. All endpoints which support this header mention this in their documentation. If an endpoint does support this header and the header is not set in the request, this value will default to the current date and time.
diff --git a/src/App.ts b/src/App.ts
index c86c66ee7c3bc5e436792ed8b6104f0b26524f4c..b23756d50d0fb23b08ceb177ab055e1e97b4d23b 100644
--- a/src/App.ts
+++ b/src/App.ts
@@ -3,6 +3,7 @@ import morgan from "morgan";
import logService from "./services/logger";
import { Connection } from "typeorm";
import MainRouter from "./router/MainRouter";
+import { Mode } from "./AppMode";
// Required for TypeORM. See https://typeorm.io/#/undefined/installation
import "reflect-metadata";
@@ -15,6 +16,7 @@ const logger = logService(module);
class App {
+ private mode: Mode
private port: number;
private connection: Connection;
private mqttClient: MqttClient;
@@ -23,7 +25,8 @@ class App {
private ingestionService: IngestionService;
private ingressService: IngressService;
- constructor(connection: Connection, mqttClient: MqttClient, channel: ConfirmChannel, port: number) {
+ constructor(mode: Mode, connection: Connection, mqttClient: MqttClient, channel: ConfirmChannel, port: number) {
+ this.mode = mode;
this.connection = connection;
this.mqttClient = mqttClient;
this.expressApp = express();
@@ -33,21 +36,49 @@ class App {
this.port = port;
}
- start() {
+ start(): void {
this.setupIngestion();
this.setupIngress();
this.setupAPI();
}
private setupIngestion() {
- this.ingestionService.start();
+ switch(this.mode) {
+ case Mode.api:
+ case Mode.ingress:
+ // Do not start the ingestion service for these modes.
+ break;
+ case Mode.development:
+ case Mode.ingestion:
+ this.ingestionService.start();
+ }
}
private setupIngress() {
- this.ingressService.start();
+ switch(this.mode) {
+ case Mode.api:
+ case Mode.ingestion:
+ // Do not start the ingress for these modes.
+ break;
+ case Mode.development:
+ case Mode.ingress:
+ this.ingressService.start();
+ }
}
private setupAPI(): void {
+ switch(this.mode) {
+ case Mode.ingestion:
+ case Mode.ingress:
+ // Do not start the api for these modes.
+ return;
+ case Mode.development:
+ case Mode.api:
+ break;
+ }
+
+ logger.info("Setting up API.");
+
// Disable ETags for this proof-of-concept to prevent some weird caching behaviour
// on some of the evaluation tools.
this.expressApp.use((req: express.Request, res: express.Response, next: express.NextFunction) => {
diff --git a/src/AppMode.ts b/src/AppMode.ts
new file mode 100644
index 0000000000000000000000000000000000000000..81e0fa50a9a592e7a6fcc45cfda91c33bd59b3ab
--- /dev/null
+++ b/src/AppMode.ts
@@ -0,0 +1,6 @@
+export enum Mode {
+ ingress,
+ ingestion,
+ api,
+ development
+}
\ No newline at end of file
diff --git a/src/config.ts b/src/config.ts
index 9115c32f8c86426abbfe0335e4ae66043282a399..a2dab20b45396d16ca26ae38168bde3f66304cc5 100644
--- a/src/config.ts
+++ b/src/config.ts
@@ -1,12 +1,13 @@
import { IClientOptions } from "mqtt";
import amqplib from "amqplib";
import logService from "./services/logger";
+import { Mode } from "./AppMode";
const logger = logService(module);
// Check if all of the required environment variables are set.
// Stop process if one of those required values is not set.
-const requiredVariables: string[] = ["DB_NAME", "API_BASE_URL"];
+const requiredVariables: string[] = ["DB_NAME", "API_BASE_URL", "MODE"];
for (const requiredVariable of requiredVariables) {
if(process.env[requiredVariable] == null) {
@@ -45,10 +46,31 @@ const rabbitMQConfig = {
queue: process.env.RABBITMQ_QUEUE ?? "measurements"
};
+let mode: Mode;
+
+switch(process.env.MODE) {
+case "development":
+ mode = Mode.development;
+ break;
+case "ingress":
+ mode = Mode.ingress;
+ break;
+case "ingestion":
+ mode = Mode.ingestion;
+ break;
+case "api":
+ mode = Mode.api;
+ break;
+default:
+ logger.error(`${process.env.MODE} is not a valid Mode.`);
+ process.exit(1);
+}
+
export default {
port: Number(process.env.API_PORT) || 8080,
baseURL: process.env.API_BASE_URL,
postgres: postgresConfig,
mqtt: mqttConfig,
- rabbitMQ: rabbitMQConfig
+ rabbitMQ: rabbitMQConfig,
+ mode: mode
};
\ No newline at end of file
diff --git a/src/index.ts b/src/index.ts
index 75385152aa50973c6e73921fd9154ec521c9e4f6..034b592acf74969348a228630b5596eb028fab93 100644
--- a/src/index.ts
+++ b/src/index.ts
@@ -31,7 +31,7 @@ dbService.then((connection) => {
mqttService.then((mqttClient) => {
if (mqttClient.connected) {
rabbitMQService.then((channel) => {
- const app = new App(connection, mqttClient, channel, config.port);
+ const app = new App(config.mode, connection, mqttClient, channel, config.port);
app.start();
}).catch((error) => {
logger.error(`Error while establishing connection to RabbitMQ: ${error}`);
diff --git a/src/services/IngestionService.ts b/src/services/IngestionService.ts
index fd51b0569b1c6640c86fdf767a7cdc8bb0a14585..736b2bfb125eee8663ed6e5db468c70bc5e7e4eb 100644
--- a/src/services/IngestionService.ts
+++ b/src/services/IngestionService.ts
@@ -62,6 +62,8 @@ class IngestionService {
}
start(): void {
+ logger.info("Starting IngestionService");
+
this.channel.consume(config.rabbitMQ.queue, (message) => {
if(!message) {
logger.error("Received empty message.");
diff --git a/src/services/IngressService.ts b/src/services/IngressService.ts
index c13eb0c1e78a807b93127bdcdee713bb56e477ba..00c1dccd8bb6eed285a8188490ee4bbdcf5c2640 100644
--- a/src/services/IngressService.ts
+++ b/src/services/IngressService.ts
@@ -16,6 +16,8 @@ export default class IngressService {
}
start(): void {
+ logger.info("Starting Ingress Service.");
+
const queue = config.rabbitMQ.queue;
this.mqttClient.on("message", (topic, message) => {