Websockets

config/config.devnet.yaml

@@ -8,7 +8,7 @@ api:
   websocket: true
 cron:
   cacheWarmer: true
-  fastWarm: true
+  fastWarm: false
   queueWorker: true
   elasticUpdater: false
 flags:
@@ -20,6 +20,9 @@ flags:
   processNfts: true
   collectionPropertiesFromGateway: false
 features:
+  websocketSubscription:
+    enabled: false
+    port: 6002
   eventsNotifier:
     enabled: false
     port: 5674
@@ -65,7 +68,7 @@ features:
     cronExpression: '*/5 * * * * *'
     activationEpoch: 1043
   chainAndromeda:
-    enabled: false
+    enabled: true
     activationEpoch: 4
   nodeEpochsLeft:
     enabled: false

config/config.e2e-mocked.mainnet.yaml

@@ -5,6 +5,9 @@ api:
   private: true
   graphql: true
 features:
+  websocketSubscription:
+    enabled: false
+    port: 6002
   dataApi:
     enabled: false
     serviceUrl: 'https://data-api.multiversx.com'

config/config.e2e.mainnet.yaml

@@ -20,6 +20,9 @@ flags:
   processNfts: true
   collectionPropertiesFromGateway: false
 features:
+  websocketSubscription:
+    enabled: false
+    port: 6002
   eventsNotifier:
     enabled: false
     port: 5674
@@ -63,7 +66,7 @@ features:
     cronExpression: '*/5 * * * * *'
     activationEpoch: 1391
   chainAndromeda:
-    enabled: false
+    enabled: true
     activationEpoch: 4
   nodeEpochsLeft:
     enabled: false

config/config.mainnet.yaml

@@ -20,6 +20,9 @@ flags:
   processNfts: true
   collectionPropertiesFromGateway: false
 features:
+  websocketSubscription:
+    enabled: false
+    port: 6002
   eventsNotifier:
     enabled: false
     port: 5674

config/config.testnet.yaml

@@ -20,6 +20,9 @@ flags:
   processNfts: true
   collectionPropertiesFromGateway: false
 features:
+  websocketSubscription:
+    enabled: false
+    port: 6002
   eventsNotifier:
     enabled: false
     port: 5674
@@ -62,7 +65,7 @@ features:
     cronExpression: '*/5 * * * * *'
     activationEpoch: 1043
   chainAndromeda:
-    enabled: false
+    enabled: true
     activationEpoch: 4
   nodeEpochsLeft:
     enabled: false

package.json

@@ -113,11 +113,13 @@
     "@sendgrid/mail": "^8.1.5",
     "agentkeepalive": "^4.2.1",
     "amqp-connection-manager": "^4.1.3",
-    "anchorme": "^3.0.8",
     "amqplib": "^0.10.0",
+    "anchorme": "^3.0.8",
     "apollo-server-core": "^3.13.0",
     "apollo-server-express": "3.13.0",
     "bignumber.js": "^9.0.2",
+    "class-transformer": "^0.5.1",
+    "class-validator": "^0.14.2",
     "compression": "^1.8.0",
     "crypto-js": "^4.1.1",
     "dataloader": "^2.2.2",
@@ -144,6 +146,8 @@
     "rxjs": "^7.1.0",
     "sharp": "^0.34.2",
     "simple-git": "^3.16.0",
+    "socket.io": "^4.8.1",
+    "socket.io-client": "^4.8.1",
     "swagger-ui-express": "^4.3.0",
     "tiny-async-pool": "^1.2.0",
     "typeorm": "^0.3.25",
@@ -216,4 +220,4 @@
       "node_modules"
     ]
   }
-}
+}

src/common/api-config/api.config.service.ts

@@ -954,4 +954,22 @@ export class ApiConfigService {
   getCompressionChunkSize(): number {
     return this.configService.get<number>('compression.chunkSize') ?? 16384;
   }
+
+  getIsWebsocketSubscriptionActive(): boolean {
+    const isWebsocketSubscriptionActive = this.configService.get<boolean>('features.websocketSubscription.enabled');
+    if (isWebsocketSubscriptionActive === undefined) {
+      throw new Error('No features.websocketSubscription.enabled flag present');
+    }
+
+    return isWebsocketSubscriptionActive;
+  }
+
+  getWebsocketSubscriptionPort(): number {
+    const port = this.configService.get<number>('features.websocketSubscription.port');
+    if (port === undefined) {
+      throw new Error('No features.websocketSubscription.port present');
+    }
+
+    return port;
+  }
 }

src/common/metrics/api.metrics.service.ts

@@ -22,6 +22,7 @@ export class ApiMetricsService {
   private static transactionsCompletedCounter: Counter<string>;
   private static transactionsPendingResultsCounter: Counter<string>;
   private static batchUpdatesCounter: Counter<string>;
+  private static subscriptionsConnectionsGauge: Gauge<string>;
 
   constructor(
     private readonly apiConfigService: ApiConfigService,
@@ -32,6 +33,13 @@ export class ApiMetricsService {
     private readonly metricsService: MetricsService,
   ) {
 
+    if (!ApiMetricsService.subscriptionsConnectionsGauge) {
+      ApiMetricsService.subscriptionsConnectionsGauge = new Gauge({
+        name: 'websocket_subscriptions_connections',
+        help: 'Number of websocket connections for subscriptions',
+      });
+    }
+
     if (!ApiMetricsService.vmQueriesHistogram) {
       ApiMetricsService.vmQueriesHistogram = new Histogram({
         name: 'vm_query',
@@ -182,6 +190,14 @@ export class ApiMetricsService {
     ApiMetricsService.lastProcessedTransactionCompletedProcessorNonce.set({ shardId }, nonce);
   }
 
+  @OnEvent(MetricsEvents.SetWebsocketMetrics)
+  setWebsocketSubscriptionsMetrics(payload: { connectedClients: number }) {
+    const { connectedClients } = payload;
+
+    ApiMetricsService.subscriptionsConnectionsGauge.set(connectedClients);
+  }
+
+
   @OnEvent(MetricsEvents.SetTransactionsCompleted)
   recordTransactionsCompleted(payload: { transactions: any[] }) {
     ApiMetricsService.transactionsCompletedCounter.inc(payload.transactions.length);