Merge pull request #26 from windingtree/feat/new-queue

New Queue library

Author: kostysh

.github/workflows/test.yml

@@ -3,6 +3,9 @@ on:
   push:
     branches:
       - develop
+  pull_request:
+    branches:
+      - develop
 jobs:
   test:
     name: Test

examples/client/src/App.tsx

@@ -45,10 +45,7 @@ export const App = () => {
   const requestsManager = useRef<
     ClientRequestsManager<RequestQuery, OfferOptions> | undefined
   >();
-  const dealsManager = useRef<ClientDealsManager<
-    RequestQuery,
-    OfferOptions
-  >>();
+  const dealsManager = useRef<ClientDealsManager<RequestQuery, OfferOptions>>();
   const { publicClient } = useWallet();
   const [connected, setConnected] = useState<boolean>(false);
   const [selectedTab, setSelectedTab] = useState<number>(0);
@@ -105,19 +102,27 @@ export const App = () => {
       );
     };
 
-    const onRequestSubscribe: EventHandler<CustomEvent<ClientRequestRecord>> = ({ detail }) => {
+    const onRequestSubscribe: EventHandler<
+      CustomEvent<ClientRequestRecord>
+    > = ({ detail }) => {
       client.current?.subscribe(detail.data.id);
     };
 
-    const onRequestUnsubscribe: EventHandler<CustomEvent<ClientRequestRecord>> = ({ detail }) => {
+    const onRequestUnsubscribe: EventHandler<
+      CustomEvent<ClientRequestRecord>
+    > = ({ detail }) => {
       client.current?.unsubscribe(detail.data.id);
     };
 
-    const onRequestPublish: EventHandler<CustomEvent<RequestData<RequestQuery>>> = ({ detail }) => {
+    const onRequestPublish: EventHandler<
+      CustomEvent<RequestData<RequestQuery>>
+    > = ({ detail }) => {
       requestsManager.current?.add(detail);
     };
 
-    const onOffer: EventHandler<CustomEvent<OfferData<RequestQuery, OfferOptions>>> = ({ detail }) => {
+    const onOffer: EventHandler<
+      CustomEvent<OfferData<RequestQuery, OfferOptions>>
+    > = ({ detail }) => {
       requestsManager.current?.addOffer(detail);
     };
 
@@ -168,8 +173,14 @@ export const App = () => {
         requestsManager.current.addEventListener('delete', updateRequests);
         requestsManager.current.addEventListener('clear', updateRequests);
         requestsManager.current.addEventListener('offer', updateRequests);
-        requestsManager.current.addEventListener('subscribe', onRequestSubscribe);
-        requestsManager.current.addEventListener('unsubscribe', onRequestUnsubscribe);
+        requestsManager.current.addEventListener(
+          'subscribe',
+          onRequestSubscribe,
+        );
+        requestsManager.current.addEventListener(
+          'unsubscribe',
+          onRequestUnsubscribe,
+        );
 
         dealsManager.current.addEventListener('changed', updateDeals);
 
@@ -200,8 +211,14 @@ export const App = () => {
       requestsManager.current?.removeEventListener('delete', updateRequests);
       requestsManager.current?.removeEventListener('clear', updateRequests);
       requestsManager.current?.removeEventListener('offer', updateRequests);
-      requestsManager.current?.removeEventListener('subscribe', onRequestSubscribe);
-      requestsManager.current?.removeEventListener('unsubscribe', onRequestUnsubscribe);
+      requestsManager.current?.removeEventListener(
+        'subscribe',
+        onRequestSubscribe,
+      );
+      requestsManager.current?.removeEventListener(
+        'unsubscribe',
+        onRequestUnsubscribe,
+      );
 
       dealsManager.current?.removeEventListener('changed', updateDeals);
 

examples/client/src/components/Deals.tsx

@@ -1,7 +1,11 @@
 import { useState, useCallback, useEffect } from 'react';
 import { DateTime } from 'luxon';
 import { Address } from 'viem';
-import { ClientDealsManager, DealRecord, DealStatus } from '../../../../src/index.js'; // @windingtree/sdk
+import {
+  ClientDealsManager,
+  DealRecord,
+  DealStatus,
+} from '../../../../src/index.js'; // @windingtree/sdk
 import { RequestQuery, OfferOptions } from '../../../shared/index.js';
 import { centerEllipsis, formatBalance, parseWalletError } from '../utils.js';
 import { useWallet } from '../providers/WalletProvider/WalletProviderContext.js';
@@ -12,27 +16,18 @@ export type DealsRegistryRecord = Required<
 
 export interface DealsProps {
   deals: DealsRegistryRecord[];
-  manager?: ClientDealsManager<
-    RequestQuery,
-    OfferOptions
-  >;
+  manager?: ClientDealsManager<RequestQuery, OfferOptions>;
 }
 
 export interface TransferFormProps {
   deal?: DealsRegistryRecord;
-  manager?: ClientDealsManager<
-    RequestQuery,
-    OfferOptions
-  >;
+  manager?: ClientDealsManager<RequestQuery, OfferOptions>;
   onClose: () => void;
 }
 
 export interface CancelProps {
   deal?: DealsRegistryRecord;
-  manager?: ClientDealsManager<
-    RequestQuery,
-    OfferOptions
-  >;
+  manager?: ClientDealsManager<RequestQuery, OfferOptions>;
   onClose: () => void;
 }
 
@@ -68,12 +63,7 @@ export const TransferForm = ({ deal, manager, onClose }: TransferFormProps) => {
         throw new Error('Ethereum client not ready');
       }
 
-      await manager.transfer(
-        deal.offer,
-        to as Address,
-        walletClient,
-        setTx,
-      );
+      await manager.transfer(deal.offer, to as Address, walletClient, setTx);
       setLoading(false);
       setSuccess(true);
     } catch (err) {

examples/client/src/components/MakeDeal.tsx

@@ -14,10 +14,7 @@ import { ConnectButton } from '../providers/WalletProvider/ConnectButton.js';
 
 interface MakeDealProps {
   offer?: OfferData<RequestQuery, OfferOptions>;
-  manager?: ClientDealsManager<
-    RequestQuery,
-    OfferOptions
-  >;
+  manager?: ClientDealsManager<RequestQuery, OfferOptions>;
 }
 
 /**
@@ -56,13 +53,7 @@ export const MakeDeal = ({ offer, manager }: MakeDealProps) => {
           throw new Error('Invalid deal configuration');
         }
 
-        await manager.create(
-          offer,
-          paymentId,
-          ZeroHash,
-          walletClient,
-          setTx,
-        );
+        await manager.create(offer, paymentId, ZeroHash, walletClient, setTx);
         setLoading(false);
         setSuccess(true);
       } catch (err) {

examples/node/index.ts

@@ -16,8 +16,8 @@ import {
   NodeOptions,
   NodeRequestManager,
   Queue,
+  JobHandler,
   createNode,
-  createJobHandler,
 } from '../../src/index.js';
 import { OfferData } from '../../src/shared/types.js';
 import { DealStatus, ProtocolContracts } from '../../src/shared/contracts.js';
@@ -63,6 +63,14 @@ process.once('unhandledRejection', (error) => {
   process.exit(1);
 });
 
+const createJobHandler =
+  <JobData = unknown, HandlerOptions = unknown>(
+    handler: JobHandler<JobData, HandlerOptions>,
+  ) =>
+  (options: HandlerOptions = {} as HandlerOptions) =>
+  (data: JobData) =>
+    handler(data, options);
+
 /**
  * This is interface of object that you want to pass to the job handler as options
  */
@@ -76,8 +84,18 @@ interface DealHandlerOptions {
 const dealHandler = createJobHandler<
   OfferData<RequestQuery, OfferOptions>,
   DealHandlerOptions
->(async ({ name, id, data: offer }, { contracts }) => {
-  logger.trace(`Job "${name}" #${id} Checking for a deal. Offer #${offer.id}`);
+>(async (offer, options) => {
+  if (!offer || !options) {
+    throw new Error('Invalid job execution configuration');
+  }
+
+  const { contracts } = options;
+
+  if (!contracts) {
+    throw new Error('Contracts manager must be provided to job handler config');
+  }
+
+  logger.trace(`Checking for a deal. Offer #${offer.id}`);
 
   // Check for a deal
   const [, , , buyer, , , status] = await contracts.getDeal(offer);
@@ -98,10 +116,10 @@ const dealHandler = createJobHandler<
       },
     );
 
-    return true; // Returning true means that the job must be stopped
+    return false; // Returning true means that the job must be stopped
   }
 
-  return; // Job continuing
+  return true; // Job continuing
 });
 
 /**
@@ -155,17 +173,20 @@ const createRequestsHandler =
         checkOut: BigInt(nowSec() + 2000),
       });
 
-      queue.addEventListener('expired', ({ detail: job }) => {
-        logger.trace(`Job #${job.id} is expired`);
+      queue.addEventListener('status', ({ detail: job }) => {
+        logger.trace(`Job #${job.id} status changed`, job);
       });
 
       /**
        * On every published offer we expecting a deal.
        * So, we add a job for detection of deals
        */
-      queue.addJob('deal', offer, {
+      queue.add({
+        handlerName: 'deal',
+        data: offer,
+        isRecurrent: true,
+        recurrenceInterval: 5000,
         expire: Number(offer.expire),
-        every: 5000, // 5 sec
       });
     };
 
@@ -211,8 +232,8 @@ const main = async (): Promise<void> => {
 
   const queue = new Queue({
     storage,
-    hashKey: 'jobs',
-    concurrentJobsNumber: 3,
+    idsKeyName: 'jobsIds',
+    concurrencyLimit: 3,
   });
 
   const requestManager = new NodeRequestManager<RequestQuery>({
@@ -234,7 +255,7 @@ const main = async (): Promise<void> => {
     requestManager.add(topic, data);
   });
 
-  queue.addJobHandler('deal', dealHandler({ contracts: contractsManager }));
+  queue.registerHandler('deal', dealHandler({ contracts: contractsManager }));
 
   /**
    * Graceful Shutdown handler

src/client/index.ts

@@ -2,6 +2,7 @@ import { EventEmitter, CustomEvent } from '@libp2p/interfaces/events';
 import { createLibp2p, Libp2pOptions, Libp2p, Libp2pInit } from 'libp2p';
 import { noise } from '@chainsafe/libp2p-noise';
 import { mplex } from '@libp2p/mplex';
+import { yamux } from '@chainsafe/libp2p-yamux';
 import { webSockets } from '@libp2p/websockets';
 import { all } from '@libp2p/websockets/filters';
 import { multiaddr, Multiaddr } from '@multiformats/multiaddr';
@@ -19,7 +20,6 @@ import { centerSub, CenterSub } from '../shared/pubsub.js';
 import { ServerAddressOption } from '../shared/options.js';
 import { encodeText, decodeText } from '../utils/text.js';
 import { createLogger } from '../utils/logger.js';
-import { yamux } from '@chainsafe/libp2p-yamux';
 
 const logger = createLogger('Client');
 

src/node/index.ts

@@ -1,6 +1,7 @@
 import { createLibp2p, Libp2pInit, Libp2pOptions, Libp2p } from 'libp2p';
 import { noise } from '@chainsafe/libp2p-noise';
 import { mplex } from '@libp2p/mplex';
+import { yamux } from '@chainsafe/libp2p-yamux';
 import { webSockets } from '@libp2p/websockets';
 import { all } from '@libp2p/websockets/filters';
 import { EventEmitter, CustomEvent } from '@libp2p/interfaces/events';
@@ -32,7 +33,6 @@ import { CenterSub, centerSub } from '../shared/pubsub.js';
 import { RequestEvent } from './requestManager.js';
 import { decodeText, encodeText } from '../utils/text.js';
 import { createLogger } from '../utils/logger.js';
-import { yamux } from '@chainsafe/libp2p-yamux';
 
 const logger = createLogger('Node');
 

src/server/index.ts

@@ -2,6 +2,7 @@ import { createLibp2p, Libp2pOptions, Libp2p } from 'libp2p';
 import { createFromJSON } from '@libp2p/peer-id-factory';
 import { noise } from '@chainsafe/libp2p-noise';
 import { mplex } from '@libp2p/mplex';
+import { yamux } from '@chainsafe/libp2p-yamux';
 import { webSockets } from '@libp2p/websockets';
 import { all } from '@libp2p/websockets/filters';
 import { EventEmitter, CustomEvent } from '@libp2p/interfaces/events';
@@ -10,7 +11,6 @@ import { centerSub, CenterSub } from '../shared/pubsub.js';
 import { decodeText } from '../utils/text.js';
 import { StorageInitializer } from '../storage/abstract.js';
 import { createLogger } from '../utils/logger.js';
-import { yamux } from '@chainsafe/libp2p-yamux';
 
 const logger = createLogger('Server');