serverTime subscription

packages/server/src/rpc/RPCRouter.ts

@@ -1,5 +1,6 @@
 import { TRPCError } from "@trpc/server"
-import { Context, Data, Effect, Layer } from "effect"
+import { observable } from "@trpc/server/observable"
+import { Context, Data, Effect, Fiber, Layer, Schedule } from "effect"
 import { TRPCBuilder } from "../trpc/TRPCBuilder"
 import { RPCProcedureBuilder } from "./procedures/RPCProcedureBuilder"
 import { todoRouter } from "./routers/todo"
@@ -14,6 +15,22 @@ export const router = Effect.gen(function*() {
             Effect.succeed("pong")
         )),
 
+        serverTime: procedure
+            .subscription(({ ctx }) =>
+                observable<string>(emit => {
+                    const emitter = ctx.fork(
+                        Effect.sync(() => emit.next(new Date().toString())).pipe(
+                            Effect.repeat(Schedule.spaced("1 second"))
+                        )
+                    )
+
+                    return () => ctx.fork(
+                        Fiber.interrupt(emitter)
+                    )
+                })
+            ),
+
+
         fail1: procedure.query(({ ctx }) => ctx.run(
             Effect.fail(new AnError({ aValue: "A value" }))
         )),

packages/webui/src/routes/index.lazy.tsx

@@ -19,17 +19,24 @@ const decodeTodos = flow(
 
 export const Index = observer(() => {
 
+    const [serverTime, setServerTime] = useState("")
+
+    trpc.serverTime.useSubscription(undefined, {
+        onData: setServerTime
+    })
+
+
     const [todos, setTodos] = useState<IObservableArray<Todo>>(observable.array())
 
     trpc.todo.changes.useSubscription(undefined, {
-        onData: data => {
+        onData: flow(decodeTodos, setTodos)
-            setTodos(decodeTodos(data))
-        }
     })
 
 
     return (
         <div className="flex flex-col gap-1 items-stretch">
+            <p className="text-center">{serverTime}</p>
+
             {todos.map(todo => (
                 <VTodo
                     key={Option.match(todo.id, {