moves poller and subscriptions controller to ws-server

Signed-off-by: Konstantina Blazhukova <konstantina.blajukova@gmail.com>

packages/relay/src/lib/relay.ts

@@ -18,12 +18,10 @@ import { DebugImpl } from './debug';
 import { EthImpl } from './eth';
 import { AdminImpl } from './admin';
 import { NetImpl } from './net';
-import { Poller } from './poller';
 import { CacheService } from './services/cacheService/cacheService';
 import HAPIService from './services/hapiService/hapiService';
 import { HbarLimitService } from './services/hbarLimitService';
 import MetricService from './services/metricService/metricService';
-import { SubscriptionController } from './subscriptionController';
 import { RequestDetails } from './types';
 import { Web3Impl } from './web3';
 
@@ -70,13 +68,6 @@ export class Relay {
    */
   private readonly ethImpl: Eth;
 
-  /**
-   * @private
-   * @readonly
-   * @property {Subs} [subImpl] - An optional implementation for handling subscription-related JSON-RPC requests.
-   */
-  private readonly subImpl?: Subs;
-
   /**
    * @private
    * @readonly
@@ -197,11 +188,6 @@ export class Relay {
       ipAddressHbarSpendingPlanRepository,
     );
 
-    if (ConfigService.get('SUBSCRIPTIONS_ENABLED')) {
-      const poller = new Poller(this.ethImpl, logger.child({ name: `poller` }), register);
-      this.subImpl = new SubscriptionController(poller, logger.child({ name: `subscr-ctrl` }), register);
-    }
-
     this.initOperatorMetric(this.clientMain, this.mirrorNodeClient, logger, register);
 
     this.populatePreconfiguredSpendingPlans().then();
@@ -292,10 +278,6 @@ export class Relay {
     return this.ethImpl;
   }
 
-  subs(): Subs | undefined {
-    return this.subImpl;
-  }
-
   mirrorClient(): MirrorNodeClient {
     return this.mirrorNodeClient;
   }

packages/ws-server/src/controllers/eth_subscribe.ts

@@ -1,19 +1,20 @@
 // SPDX-License-Identifier: Apache-2.0
 
+import { ConfigService } from '@hashgraph/json-rpc-config-service/dist/services';
 import { predefined, Relay } from '@hashgraph/json-rpc-relay/dist';
-import { validateSubscribeEthLogsParams } from '../utils/validators';
-import constants from '@hashgraph/json-rpc-relay/dist/lib/constants';
 import { MirrorNodeClient } from '@hashgraph/json-rpc-relay/dist/lib/clients';
-import jsonResp from '@hashgraph/json-rpc-server/dist/koaJsonRpc/lib/RpcResponse';
-import { constructValidLogSubscriptionFilter, getMultipleAddressesEnabled } from '../utils/utils';
-import { ConfigService } from '@hashgraph/json-rpc-config-service/dist/services';
+import constants from '@hashgraph/json-rpc-relay/dist/lib/constants';
 import { RequestDetails } from '@hashgraph/json-rpc-relay/dist/lib/types';
-import { ISharedParams } from './index';
 import { IJsonRpcRequest } from '@hashgraph/json-rpc-server/dist/koaJsonRpc/lib/IJsonRpcRequest';
 import { IJsonRpcResponse } from '@hashgraph/json-rpc-server/dist/koaJsonRpc/lib/IJsonRpcResponse';
+import jsonResp from '@hashgraph/json-rpc-server/dist/koaJsonRpc/lib/RpcResponse';
 import { Context } from 'koa';
 import { Logger } from 'pino';
 
+import { getSubscriptionController } from '../subscriptionsManager';
+import { constructValidLogSubscriptionFilter, getMultipleAddressesEnabled } from '../utils/utils';
+import { validateSubscribeEthLogsParams } from '../utils/validators';
+import { ISharedParams } from './index';
 /**
  * Subscribes to new block headers (newHeads) events and returns the response and subscription ID.
  * @param {any} filters - The filters object specifying criteria for the subscription.
@@ -23,16 +24,12 @@ import { Logger } from 'pino';
  * @param {Logger} logger - The logger object used for logging subscription information.
  * @returns {string | undefined} Returns the subscription ID.
  */
-const subscribeToNewHeads = (
-  filters: any,
-  ctx: Context,
-  event: string,
-  relay: Relay,
-  logger: Logger,
-): string | undefined => {
-  const subscriptionId = relay.subs()?.subscribe(ctx.websocket, event, filters);
-  logger.info(`Subscribed to newHeads, subscriptionId: ${subscriptionId}`);
-  return subscriptionId;
+const subscribeToNewHeads = (filters: any, ctx: Context, event: string, logger: Logger): string | undefined => {
+  const subscriptionController = getSubscriptionController();
+
+  const id = subscriptionController?.subscribe(ctx.websocket, event, filters);
+  logger.info(`Subscribed to newHeads, subscriptionId: ${id}`);
+  return id;
 };
 
 /**
@@ -65,7 +62,7 @@ const handleEthSubscribeNewHeads = (
     throw predefined.UNSUPPORTED_METHOD;
   }
 
-  const subscriptionId = subscribeToNewHeads(filters, ctx, event, relay, logger);
+  const subscriptionId = subscribeToNewHeads(filters, ctx, event, logger);
   return jsonResp(request.id, null, subscriptionId);
 };
 
@@ -101,8 +98,8 @@ const handleEthSubscribeLogs = async (
   ) {
     throw predefined.INVALID_PARAMETER('filters.address', 'Only one contract address is allowed');
   }
-
-  const subscriptionId = relay.subs()?.subscribe(ctx.websocket, event, validFiltersObject);
+  const subscriptionController = getSubscriptionController();
+  const subscriptionId = subscriptionController?.subscribe(ctx.websocket, event, validFiltersObject);
   return jsonResp(request.id, null, subscriptionId);
 };
 
@@ -151,21 +148,3 @@ export const handleEthSubscribe = async ({
 
   return response;
 };
-
-/**
- * Handles unsubscription requests for on-chain events.
- * Unsubscribes the WebSocket from the specified subscription ID and returns the response.
- * @param {object} args - An object containing the function parameters as properties.
- * @param {Context} args.ctx - The context object containing information about the WebSocket connection.
- * @param {any[]} args.params - The parameters of the unsubscription request.
- * @param {IJsonRpcRequest} args.request - The request object received from the client.
- * @param {Relay} args.relay - The relay object used for managing WebSocket subscriptions.
- * @param {ConnectionLimiter} args.limiter - The limiter object used for rate limiting WebSocket connections.
- * @returns {IJsonRpcResponse} Returns the response to the unsubscription request.
- */
-export const handleEthUnsubscribe = ({ ctx, params, request, relay, limiter }: ISharedParams): IJsonRpcResponse => {
-  const subId = params[0];
-  const unsubbedCount = relay.subs()?.unsubscribe(ctx.websocket, subId);
-  limiter.decrementSubs(ctx, unsubbedCount);
-  return jsonResp(request.id, null, unsubbedCount !== 0);
-};

packages/ws-server/src/controllers/eth_unsubscribe.ts

@@ -0,0 +1,25 @@
+// SPDX-License-Identifier: Apache-2.0
+import { IJsonRpcResponse } from '@hashgraph/json-rpc-server/dist/koaJsonRpc/lib/IJsonRpcResponse';
+import jsonResp from '@hashgraph/json-rpc-server/dist/koaJsonRpc/lib/RpcResponse';
+
+import { getSubscriptionController } from '../subscriptionsManager';
+import { ISharedParams } from '.';
+
+/**
+ * Handles unsubscription requests for on-chain events.
+ * Unsubscribes the WebSocket from the specified subscription ID and returns the response.
+ * @param {object} args - An object containing the function parameters as properties.
+ * @param {Context} args.ctx - The context object containing information about the WebSocket connection.
+ * @param {any[]} args.params - The parameters of the unsubscription request.
+ * @param {IJsonRpcRequest} args.request - The request object received from the client.
+ * @param {Relay} args.relay - The relay object used for managing WebSocket subscriptions.
+ * @param {ConnectionLimiter} args.limiter - The limiter object used for rate limiting WebSocket connections.
+ * @returns {IJsonRpcResponse} Returns the response to the unsubscription request.
+ */
+export const handleEthUnsubscribe = ({ ctx, params, request, limiter }: ISharedParams): IJsonRpcResponse => {
+  const subscriptionsController = getSubscriptionController();
+  const subId = params[0];
+  const unsubbedCount = subscriptionsController?.unsubscribe(ctx.websocket, subId);
+  limiter.decrementSubs(ctx, unsubbedCount);
+  return jsonResp(request.id, null, unsubbedCount !== 0);
+};

packages/ws-server/src/controllers/index.ts

@@ -1,24 +1,26 @@
 // SPDX-License-Identifier: Apache-2.0
 
-import { WS_CONSTANTS } from '../utils/constants';
-import WsMetricRegistry from '../metrics/wsMetricRegistry';
-import ConnectionLimiter from '../metrics/connectionLimiter';
-import { Validator } from '@hashgraph/json-rpc-server/dist/validator';
-import { handleEthSubscribe, handleEthUnsubscribe } from './eth_subscribe';
 import { JsonRpcError, predefined, Relay } from '@hashgraph/json-rpc-relay/dist';
 import { MirrorNodeClient } from '@hashgraph/json-rpc-relay/dist/lib/clients';
-import jsonResp from '@hashgraph/json-rpc-server/dist/koaJsonRpc/lib/RpcResponse';
-import { paramRearrangementMap, validateJsonRpcRequest, verifySupportedMethod } from '../utils/utils';
+import { RequestDetails } from '@hashgraph/json-rpc-relay/dist/lib/types';
+import { IJsonRpcRequest } from '@hashgraph/json-rpc-server/dist/koaJsonRpc/lib/IJsonRpcRequest';
+import { IJsonRpcResponse } from '@hashgraph/json-rpc-server/dist/koaJsonRpc/lib/IJsonRpcResponse';
 import {
   InvalidRequest,
   IPRateLimitExceeded,
   MethodNotFound,
 } from '@hashgraph/json-rpc-server/dist/koaJsonRpc/lib/RpcError';
-import { RequestDetails } from '@hashgraph/json-rpc-relay/dist/lib/types';
-import { Logger } from 'pino';
-import { IJsonRpcRequest } from '@hashgraph/json-rpc-server/dist/koaJsonRpc/lib/IJsonRpcRequest';
+import jsonResp from '@hashgraph/json-rpc-server/dist/koaJsonRpc/lib/RpcResponse';
+import { Validator } from '@hashgraph/json-rpc-server/dist/validator';
 import Koa from 'koa';
-import { IJsonRpcResponse } from '@hashgraph/json-rpc-server/dist/koaJsonRpc/lib/IJsonRpcResponse';
+import { Logger } from 'pino';
+
+import ConnectionLimiter from '../metrics/connectionLimiter';
+import WsMetricRegistry from '../metrics/wsMetricRegistry';
+import { WS_CONSTANTS } from '../utils/constants';
+import { paramRearrangementMap, validateJsonRpcRequest, verifySupportedMethod } from '../utils/utils';
+import { handleEthSubscribe } from './eth_subscribe';
+import { handleEthUnsubscribe } from './eth_unsubscribe';
 
 export type ISharedParams = {
   request: IJsonRpcRequest;
@@ -110,6 +112,7 @@ export const getRequestResult = async (
   requestDetails: RequestDetails,
 ): Promise<any> => {
   // Extract the method and parameters from the received request
+  // eslint-disable-next-line
   let { method, params } = request;
 
   // support go-ethereum client by turning undefined into empty array

packages/relay/src/lib/subscriptionController.ts → packages/ws-server/src/controllers/subscriptionController.ts

@@ -1,15 +1,14 @@
 // SPDX-License-Identifier: Apache-2.0
 
 import { ConfigService } from '@hashgraph/json-rpc-config-service/dist/services';
+import { generateRandomHex } from '@hashgraph/json-rpc-relay/dist/formatters';
 import crypto from 'crypto';
+import { Context } from 'koa';
 import LRU from 'lru-cache';
 import { Logger } from 'pino';
 import { Counter, Histogram, Registry } from 'prom-client';
 
-import { generateRandomHex } from '../formatters';
-import { Subs } from '../index';
-import constants from './constants';
-import { Poller } from './poller';
+import { Poller } from '../poller';
 
 export interface Subscriber {
   connection: any;
@@ -19,7 +18,7 @@ export interface Subscriber {
 
 const CACHE_TTL = ConfigService.get('WS_CACHE_TTL');
 
-export class SubscriptionController implements Subs {
+export class SubscriptionController {
   private poller: Poller;
   private logger: Logger;
   private subscriptions: { [key: string]: Subscriber[] };
@@ -63,16 +62,16 @@ export class SubscriptionController implements Subs {
     });
   }
 
-  createHash(data) {
+  private createHash(data: string) {
     return crypto.createHash('sha256').update(data.toString()).digest('hex');
   }
 
   // Generates a random 16 byte hex string
-  generateId() {
+  public generateId() {
     return generateRandomHex();
   }
 
-  subscribe(connection, event: string, filters?: {}) {
+  public subscribe(connection: Context, event: string, filters?: {}) {
     let tag: any = { event };
     if (filters && Object.keys(filters).length) {
       tag.filters = filters;
@@ -110,7 +109,7 @@ export class SubscriptionController implements Subs {
     return subId;
   }
 
-  unsubscribe(connection, subId?: string) {
+  public unsubscribe(connection, subId?: string) {
     const { id } = connection;
 
     if (subId) {
@@ -148,7 +147,7 @@ export class SubscriptionController implements Subs {
     return subCount;
   }
 
-  notifySubscribers(tag, data) {
+  public notifySubscribers(tag, data) {
     if (this.subscriptions[tag] && this.subscriptions[tag].length) {
       this.subscriptions[tag].forEach((sub) => {
         const subscriptionData = {