feat: wip

.github/workflows/main.yml

@@ -36,7 +36,7 @@ jobs:
         run: yarn build
 
       - name: Setup example
-        run: cd example/next-live-example && yarn install --frozen-lockfile && yarn link next-live && yarn build
+        run: cd example/next-realtime-example && yarn install --frozen-lockfile && yarn link next-realtime && yarn build
 
       - name: Check types
         run: yarn type-check

examples/next-live-example/src/actions/addTodo.ts → examples/next-realtime-example/src/actions/addTodo.ts

@@ -1,7 +1,7 @@
 'use server';
 
 import { nanoid } from 'nanoid';
-import { revalidateLiveTag } from '../app/next-live/config';
+import { revalidateRealtimeTag } from '../app/realtime/config';
 import { getDb } from '../drizzle/getDb';
 import { todoTable } from '../drizzle/schema';
 
@@ -17,5 +17,5 @@ export const addTodo = async ({ text }: { text: string }) => {
     text: text.trim(),
   });
 
-  await revalidateLiveTag('todos');
+  await revalidateRealtimeTag('todos');
 };

examples/next-live-example/src/actions/clearTodos.ts → examples/next-realtime-example/src/actions/clearTodos.ts

@@ -1,6 +1,6 @@
 'use server';
 
-import { revalidateLiveTag } from '../app/next-live/config';
+import { revalidateRealtimeTag } from '../app/realtime/config';
 import { getDb } from '../drizzle/getDb';
 import { todoTable } from '../drizzle/schema';
 
@@ -9,5 +9,5 @@ export const clearTodos = async () => {
 
   await db.delete(todoTable).execute();
 
-  await revalidateLiveTag('todos');
+  await revalidateRealtimeTag('todos');
 };

examples/next-live-example/src/actions/deleteTodo.ts → examples/next-realtime-example/src/actions/deleteTodo.ts

@@ -1,7 +1,7 @@
 'use server';
 
 import { eq } from 'drizzle-orm';
-import { revalidateLiveTag } from '../app/next-live/config';
+import { revalidateRealtimeTag } from '../app/realtime/config';
 import { getDb } from '../drizzle/getDb';
 import { todoTable } from '../drizzle/schema';
 
@@ -14,5 +14,5 @@ export const deleteTodo = async ({ id }: { id: string }) => {
 
   await db.delete(todoTable).where(eq(todoTable.id, id));
 
-  await revalidateLiveTag('todos');
+  await revalidateRealtimeTag('todos');
 };

examples/next-live-example/src/app/page.tsx → examples/next-realtime-example/src/app/page.tsx

@@ -1,4 +1,4 @@
-import { NextLiveStream } from 'next-live/client';
+import { NextRealtimeStream } from 'next-realtime/client';
 import { revalidateTag } from 'next/cache';
 import TodoList from '../components/TodoList';
 
@@ -7,7 +7,7 @@ export const dynamic = 'force-dynamic';
 export default async function Home() {
   return (
     <>
-      <NextLiveStream
+      <NextRealtimeStream
         type="redis"
         revalidateTag={async (tag: string) => {
           'use server';

examples/next-realtime-example/src/app/realtime/config.ts

@@ -0,0 +1,9 @@
+import Redis from 'ioredis';
+import { configNextRealtimePostgres } from 'next-realtime';
+import { client } from '../../drizzle/getDbServer';
+
+export const redis = new Redis(process.env.REDIS_CONNECTION_STRING as string);
+
+export const { NextRealtimeStreamHandler, revalidateRealtimeTag } =
+  // configNextRealtimeRedis(redis);
+  configNextRealtimePostgres(client);

examples/next-realtime-example/src/app/realtime/route.ts

@@ -0,0 +1,5 @@
+import { NextRealtimeStreamHandler } from './config';
+
+export const dynamic = 'force-dynamic';
+
+export const GET = NextRealtimeStreamHandler;

examples/next-live-example/src/drizzle/getDbServer.ts → examples/next-realtime-example/src/drizzle/getDbServer.ts

@@ -3,13 +3,14 @@ import postgres from 'postgres';
 
 let db: ReturnType<typeof drizzle>;
 
+// eslint-disable-next-line no-underscore-dangle
+export const client = postgres(
+  `${process.env.PG_CONNECTION_STRING}/${process.env.PG_DB}`
+);
+
 export const getDbServer = () => {
   if (!db) {
-    // eslint-disable-next-line no-underscore-dangle
-    const _client = postgres(
-      `${process.env.PG_CONNECTION_STRING}/${process.env.PG_DB}`
-    );
-    db = drizzle(_client);
+    db = drizzle(client);
   }
 
   return db;

package.json

@@ -1,5 +1,5 @@
 {
-  "name": "next-live",
+  "name": "next-realtime",
   "description": "",
   "version": "0.0.0",
   "author": "",

src/client/index.ts

@@ -3,13 +3,13 @@ import { useCallback, useState, useTransition, useEffect, useRef } from 'react';
 // @ts-ignore
 import { useStream } from 'react-fetch-streams';
 
-export interface NextLiveProps {
+export interface NextRealtimeProps {
   baseUrl?: string;
   path?: string;
   revalidateTag: (tag: string) => Promise<void>;
 }
 
-export const NextLiveStream = (props: NextLiveProps) => {
+export const NextRealtimeStream = (props: NextRealtimeProps) => {
   const [, startTransition] = useTransition();
 
   const onNext = useCallback(async (res: any) => {
@@ -33,7 +33,7 @@ export const NextLiveStream = (props: NextLiveProps) => {
     // console.log('error', error);
   }, []);
 
-  const url = `${props.baseUrl || ''}/${props.path || `next-live`}`;
+  const url = `${props.baseUrl || ''}/${props.path || `realtime`}`;
 
   const [id, setId] = useState(nanoid());
 
@@ -51,19 +51,19 @@ export const NextLiveStream = (props: NextLiveProps) => {
   return null;
 };
 
-export interface NextLivePollingProps extends NextLiveProps {
+export interface NextRealtimePollingProps extends NextRealtimeProps {
   pollInterval?: number;
 }
 
-export const NextLivePolling = (props: NextLivePollingProps) => {
+export const NextRealtimePolling = (props: NextRealtimePollingProps) => {
   const isPageVisible = usePageVisibility();
   const timerIdRef = useRef(null);
   const [isPollingEnabled, setIsPollingEnabled] = useState(true);
 
   useEffect(() => {
     const pollingCallback = async () => {
       const res = await fetch(
-        `${props.baseUrl || ''}/${props.path || `next-live`}`
+        `${props.baseUrl || ''}/${props.path || `realtime`}`
       );
 
       if (res.status === 200) {

src/index.ts

@@ -8,65 +8,74 @@ export type SubscribeFn = (
   callback: () => void
 ) => Promise<void>;
 
-export type OnMessageFn = (callback: (message: any) => void) => Promise<void>;
+export type OnMessageFn = (
+  channel: string,
+  enqueue: (message: any) => void
+) => Promise<void>;
 
-export const configNextLive = (options: {
+export const configNextRealtime = (options: {
   onMessage: OnMessageFn;
   publish: PublishFn;
   subscribe: SubscribeFn;
 }) => {
-  const revalidateLiveTag = (tags: string[] | string = []) => {
+  const revalidateRealtimeTag = (tags: string[] | string = []) => {
     const _tags = Array.isArray(tags) ? tags : [tags];
     _tags.forEach((tag) => {
       revalidateTag(tag);
     });
-    return options.publish('next-live', JSON.stringify({ tags: _tags }));
+    return options.publish('next-realtime', JSON.stringify({ tags: _tags }));
   };
 
-  const NextLiveStreamHandler = () => {
+  const NextRealtimeStreamHandler = () => {
     const encoder = new TextEncoder();
     const stream = new ReadableStream({
       async start(controller) {
-        await options.subscribe('next-live', () => {});
+        await options.subscribe('next-realtime', () => {});
         const enqueue = (message: any) => {
           try {
             controller.enqueue(encoder.encode(message));
           } catch (e) {
             //
           }
         };
-        await options.onMessage(enqueue);
-      },
-    });
-    return new Response(stream, {
-      headers: {
-        'Content-Type': 'text/event-stream; charset=utf-8',
-        Connection: 'keep-alive',
-        'Cache-Control': 'no-cache, no-transform',
-        'Content-Encoding': 'none',
+        await options.onMessage('next-realtime', enqueue);
       },
     });
+    return new Response(stream);
   };
 
-  const NextLivePollingHandler = () => {};
+  const NextRealtimePollingHandler = () => {};
 
   return {
-    NextLiveStreamHandler,
-    NextLivePollingHandler,
-    revalidateLiveTag,
+    NextRealtimeStreamHandler,
+    NextRealtimePollingHandler,
+    revalidateRealtimeTag,
   };
 };
 
-export const configNextLiveRedis = (redis: any) =>
-  configNextLive({
+export const configNextRealtimeRedis = (redis: any) =>
+  configNextRealtime({
     publish: async (channel: any, message: any) => {
       await redis.publish(channel, message);
     },
     subscribe: async (channel: any, callback: any) => {
       await redis.subscribe(channel, callback);
     },
-    onMessage: async (enqueue: any) => {
-      await redis.on('message', (channel: string, message: string) => {
+    onMessage: async (channel: any, enqueue: any) => {
+      await redis.on('message', (_channel: string, message: string) => {
+        enqueue(message);
+      });
+    },
+  });
+
+export const configNextRealtimePostgres = (client: any) =>
+  configNextRealtime({
+    publish: async (channel: any, message: any) => {
+      await client.notify(channel, message);
+    },
+    subscribe: async () => {},
+    onMessage: async (channel, enqueue: any) => {
+      await client.listen(channel, (message: string) => {
         enqueue(message);
       });
     },