refactor(./class/stream/~): use existing instance (#170)

* refactor(./class/stream/~): use existing instance * doc(./stream/~): updated doc * test(./test/class/stream/~): updated UT
MyUnisoft · Jan 31, 2025 · 9e55b28 · 9e55b28
1 parent 40014f3
commit 9e55b28
Show file tree

Hide file tree

Showing 11 changed files with 113 additions and 88 deletions.
diff --git a/docs/pubsub/Channel.md b/docs/pubsub/Channel.md
@@ -27,7 +27,7 @@ export type PublishOptions<
 ## 📚 Usage
 
 ```ts
-import { Channel } from "@myunisoft/redis";
+import { Channel, RedisAdapter } from "@myunisoft/redis";
 
 const name = "foo";
 const redis = new RedisAdapter({
@@ -40,6 +40,7 @@ const subscriber = new RedisAdapter({
 });
 
 await redis.initialize();
+await subscriber.initialize();
 
 await subscriber.subscribe(name);
 subscriber.on("message", (channel, message) => {

diff --git a/docs/stream/Interpersonal.md b/docs/stream/Interpersonal.md
@@ -16,7 +16,7 @@ interface ClaimOptions {
   idleTime: number;
 }
 
-interface GroupConsumerOptions extends BasementOptions {
+export interface InterpersonalOptions extends StreamOptions {
   groupName: string;
   consumerName: string;
   claimOptions?: ClaimOptions;
@@ -26,10 +26,17 @@ interface GroupConsumerOptions extends BasementOptions {
 ## 📚 Usage
 
 ```ts
-import { GroupConsumer } from "@myunisoft/redis";
+import { Interpersonal, RedisAdapter } from "@myunisoft/redis";
 
-const consumer = new GroupConsumer({
-  connection,
+const redis = new RedisAdapter({
+  port: Number(process.env.REDIS_PORT),
+  host: process.env.REDIS_HOST
+});
+
+await redis.initialize();
+
+const consumer = new Interpersonal({
+  redis,
   streamName: "my-stream-name",
   groupName: "my-group-name",
   consumerName: "my-consumer-name",
@@ -41,7 +48,6 @@ const consumer = new GroupConsumer({
   }
 });
 
-await consumer.initialize();
 await consumer.init();
 
 const readable = Readable.from(firstConsumer[Symbol.asyncIterator]());

diff --git a/docs/stream/Intrapersonal.md b/docs/stream/Intrapersonal.md
@@ -10,17 +10,23 @@
 ## 📚 Usage
 
 ```ts
-import { Intrapersonal } from "@myunisoft/redis";
+import { Intrapersonal, RedisAdapter } from "@myunisoft/redis";
+
+const redis = new RedisAdapter({
+  port: Number(process.env.REDIS_PORT),
+  host: process.env.REDIS_HOST
+});
+
+await redis.initialize();
 
 const consumer = new Intrapersonal({
-  connection,
+  redis,
   streamName: "my-stream-name",
   frequency: 10000, 
   lastId: "0-0",
   count: 10
 });
 
-await consumer.initialize();
 await consumer.init();
 
 const readable = Readable.from(basicStream[Symbol.asyncIterator]());

diff --git a/docs/stream/Stream.md b/docs/stream/Stream.md
@@ -65,17 +65,24 @@ interface ConsumeOptions {
 ## 📚 Usage
 
 ```ts
-import { Stream } from "@myunisoft/redis";
+import { Stream, RedisAdapter } from "@myunisoft/redis";
 
-const redisStream = new Stream({
+const redis = new RedisAdapter({
+  port: Number(process.env.REDIS_PORT),
+  host: process.env.REDIS_HOST
+});
+
+await redis.initialize();
+
+const stream = new Stream({
+  redis,
   streamName: "my-stream-name",
   frequency: 10000, 
   lastId: "0-0",
   count: 10
 });
 
-await redisStream.initialize();
-await redisStream.init();
+await stream.init();
 ```
 
 ## 📜 API

diff --git a/src/class/stream/Interpersonal.class.ts b/src/class/stream/Interpersonal.class.ts
@@ -102,7 +102,7 @@ export class Interpersonal extends Stream {
         ...args: RedisValue[]
       ];
 
-      streamResults = await this.xreadgroup(...optionsWithCount);
+      streamResults = await this.redis.xreadgroup(...optionsWithCount);
     }
     else {
       const optionsWithoutCount = [...redisOptions] as [
@@ -113,7 +113,7 @@ export class Interpersonal extends Stream {
         ...args: RedisValue[]
       ];
 
-      streamResults = await this.xreadgroup(...optionsWithoutCount);
+      streamResults = await this.redis.xreadgroup(...optionsWithoutCount);
     }
 
     if (!streamResults) {
@@ -144,7 +144,7 @@ export class Interpersonal extends Stream {
         count: number | string
       ];
 
-      streamResults = await this.xautoclaim(...optionsWithCount);
+      streamResults = await this.redis.xautoclaim(...optionsWithCount);
     }
     else {
       const optionsWithoutCount = [...redisOptions] as [
@@ -155,29 +155,25 @@ export class Interpersonal extends Stream {
         start: RedisKey | number
       ];
 
-      streamResults = await this.xautoclaim(...optionsWithoutCount);
+      streamResults = await this.redis.xautoclaim(...optionsWithoutCount);
     }
 
-    const [cursor, entries] = streamResults;
+    const [__, entries] = streamResults;
 
     if (entries.length === 0) {
       return [];
     }
 
-    if (cursor !== "0-0") {
-      this.emit("rest");
-    }
-
     return await this.handleEntries(entries, ">");
   }
 
   public async claimEntry(entryId: string): Promise<void> {
-    await this.xack(this.streamName, this.groupName, entryId);
+    await this.redis.xack(this.streamName, this.groupName, entryId);
     await this.delEntry(entryId);
   }
 
   override async getConsumerData(): Promise<utils.XINFOConsumerData | undefined> {
-    const consumers = await this.xinfo("CONSUMERS", this.streamName, this.groupName);
+    const consumers = await this.redis.xinfo("CONSUMERS", this.streamName, this.groupName);
 
     const formattedConsumers = utils.parseXINFOConsumers(consumers as utils.XINFOConsumers);
 
@@ -196,7 +192,7 @@ export class Interpersonal extends Stream {
       return;
     }
 
-    await this.xgroup("CREATE", this.streamName, this.groupName, "$", "MKSTREAM");
+    await this.redis.xgroup("CREATE", this.streamName, this.groupName, "$", "MKSTREAM");
   }
 
   override async deleteGroup() {
@@ -205,7 +201,7 @@ export class Interpersonal extends Stream {
       return;
     }
 
-    await this.xgroup("DESTROY", this.streamName, this.groupName);
+    await this.redis.xgroup("DESTROY", this.streamName, this.groupName);
   }
 
   override async consumerExist(): Promise<boolean> {
@@ -220,7 +216,7 @@ export class Interpersonal extends Stream {
       return;
     }
 
-    await this.xgroup("CREATECONSUMER", this.streamName, this.groupName, this.consumerName);
+    await this.redis.xgroup("CREATECONSUMER", this.streamName, this.groupName, this.consumerName);
   }
 
   override async deleteConsumer(): Promise<void> {
@@ -229,6 +225,6 @@ export class Interpersonal extends Stream {
       return;
     }
 
-    await this.xgroup("DELCONSUMER", this.streamName, this.groupName, this.consumerName);
+    await this.redis.xgroup("DELCONSUMER", this.streamName, this.groupName, this.consumerName);
   }
 }
diff --git a/src/class/stream/Intrapersonal.class.ts b/src/class/stream/Intrapersonal.class.ts
@@ -38,7 +38,7 @@ export class Intrapersonal extends Stream {
         ...args: RedisValue[]
       ];
 
-      streamResults = await this.xread(...optionsWithCount);
+      streamResults = await this.redis.xread(...optionsWithCount);
     }
     else if (block) {
       const optionsWithBlock = [...redisOptions] as [
@@ -48,15 +48,15 @@ export class Intrapersonal extends Stream {
         ...args: RedisValue[]
       ];
 
-      streamResults = await this.xread(...optionsWithBlock);
+      streamResults = await this.redis.xread(...optionsWithBlock);
     }
     else {
       const optionsWithoutCount = [...redisOptions] as [
         streamsToken: "STREAMS",
         ...args: RedisValue[]
       ];
 
-      streamResults = await this.xread(...optionsWithoutCount);
+      streamResults = await this.redis.xread(...optionsWithoutCount);
     }
 
     if (!streamResults) {