chore: misc cleanup (#267)

_tasks/download_frame_metadata.ts

@@ -3,14 +3,14 @@ import * as path from "../deps/std/path.ts";
 import { assert } from "../deps/std/testing/asserts.ts";
 import { acala, kusama, moonbeam, polkadot, statemint, subsocial, westend } from "../known/mod.ts";
 import * as rpc from "../rpc/mod.ts";
+import * as U from "../util/mod.ts";
 
 const outDir = path.join(Deno.cwd(), "frame_metadata", "_downloaded");
 await fs.emptyDir(outDir);
 await Promise.all(
   Object.entries({ acala, kusama, moonbeam, polkadot, statemint, subsocial, westend }).map(
     async ([name, config]) => {
-      const client = await rpc.stdClient(config);
-      assert(!(client instanceof Error));
+      const client = U.throwIfError(await rpc.stdClient(config));
       try {
         const metadata = await client.call("state_getMetadata", []);
         assert(metadata.result);

effect/std/watchBlocks.ts

@@ -21,11 +21,10 @@ export function watchBlocks(
       const blockHash = a
         .rpcCall(config, "chain_getBlockHash", [blockNum])
         .select("result");
-      const block = await readBlock(config, blockHash as unknown as U.HashHexString).run(); // STOP THIS MADNESS
-      if (block instanceof Error) {
-        // TODO: subscription runtime error channel
-        throw block;
-      }
+      const block = U.throwIfError(
+        // STOP THIS MADNESS
+        await readBlock(config, blockHash as unknown as U.HashHexString).run(),
+      );
       watchHandler(block.block);
     };
   }, (ok) => {

examples/watch_blocks_iter.ts

@@ -0,0 +1,20 @@
+// TODO: Decide whether we want to make anything of the following DX.
+//       The main issue here is that the `T` of `watchIter` is `unknown`.
+//       We could fix this by turning `watchBlocks`'s `run` into an assertion.
+//       **Dangerous**.
+
+import * as C from "../mod.ts";
+
+const watchIter = C.watchIter();
+
+C.watchBlocks(C.rococo, watchIter).run();
+
+let i = 0;
+
+for await (const block of watchIter) {
+  console.log({ [i]: block });
+  if (i === 5) {
+    break;
+  }
+  i++;
+}

mod.ts

@@ -5,4 +5,11 @@ export * from "./effect/mod.ts";
 export * as M from "./frame_metadata/mod.ts";
 export { $era, $null, ChainError, type Era } from "./frame_metadata/mod.ts";
 export { kusama, moonbeam, polkadot, rococo, westend } from "./known/mod.ts"; // TODO: get rid of this!
-export { type Handle, handle, hex, Iter, mapCreateWatchHandler } from "./util/mod.ts";
+export {
+  type CreateWatchHandler,
+  hex,
+  mapCreateWatchHandler,
+  type WatchHandler,
+  type WatchIter,
+  watchIter,
+} from "./util/mod.ts";

rpc/providers/proxy.test.ts

@@ -1,5 +1,6 @@
 import { assert } from "../../deps/std/testing/asserts.ts";
 import { polkadot } from "../../known/mod.ts";
+import * as U from "../../util/mod.ts";
 import * as msg from "../messages.ts";
 import { proxyClient } from "./proxy.ts";
 
@@ -8,8 +9,7 @@ Deno.test({
   sanitizeResources: false,
   sanitizeOps: false,
   async fn(t) {
-    const client = await proxyClient(polkadot);
-    assert(!(client instanceof Error));
+    const client = U.throwIfError(await proxyClient(polkadot));
 
     await t.step("call", async () => {
       const raw = await client.call("state_getMetadata", []);

rpc/providers/std.ts

@@ -1,6 +1,5 @@
 import { Config } from "../../config/mod.ts";
 import { unimplemented } from "../../deps/std/testing/asserts.ts";
-import { isWsUrl } from "../../util/mod.ts";
 import { FailedToOpenConnectionError, ProxyClient, proxyClient } from "./proxy.ts";
 import {
   FailedToAddChainError,
@@ -22,7 +21,8 @@ export function stdClient<Config_ extends Config<string>>(
   config: Config_,
 ): Promise<StdClient<Config_> | StdClientInitError> {
   if (typeof config.discoveryValue === "string") {
-    if (isWsUrl(config.discoveryValue)) {
+    // TODO: improve check / move selection elsewhere
+    if (config.discoveryValue.startsWith("ws")) {
       return proxyClient(config);
     } else {
       return smoldotClient(config);

util/mod.ts

@@ -1,7 +1,5 @@
 export * from "./branded.ts";
-export * from "./discovery_value_validation.ts";
 export * from "./error.ts";
 export * as hex from "./hex.ts";
-export * from "./iter.ts";
 export * from "./types.ts";
 export * from "./watch.ts";

util/types.ts

@@ -2,11 +2,3 @@ export type ValueOf<T> = T[keyof T];
 
 export type U2I<T> = (T extends any ? (x: T) => any : never) extends (x: infer R) => any ? R
   : never;
-
-export type EnsureLookup<
-  K extends PropertyKey,
-  ValueConstraint,
-  Lookup extends {
-    [_ in K]: ValueConstraint;
-  },
-> = Lookup;

util/watch.ts

@@ -1,7 +1,7 @@
-import { Iter } from "./iter.ts";
+import { deferred } from "../deps/std/async.ts";
 
 export type WatchHandler<Event> = (event: Event) => void;
-export type CreateWatchHandler<IngressMessage> = (stop: () => void) => WatchHandler<IngressMessage>;
+export type CreateWatchHandler<Event> = (stop: () => void) => WatchHandler<Event>;
 
 export function mapCreateWatchHandler<From, Into>(
   createWatchHandler: CreateWatchHandler<Into>,
@@ -15,14 +15,48 @@ export function mapCreateWatchHandler<From, Into>(
   };
 }
 
-// TODO: rename
-export type Handle<T> = CreateWatchHandler<T> & { iter: Iter<T> };
-export function handle<T>(): Handle<T> {
-  const inner = new Iter<T>();
-  const createWatchHandler: Handle<T> = (stop) => {
-    inner.onDone = stop;
-    return inner.push;
+export type WatchIter<T> = CreateWatchHandler<T> & AsyncIterable<T>;
+
+export function watchIter<T>(): WatchIter<T> {
+  const queue: T[] = [];
+  const cbs: ((value: IteratorYieldResult<T>) => void)[] = [];
+  const onDoneContainer: { onDone?: () => void } = {};
+  const createWatchHandler: CreateWatchHandler<T> = (stop) => {
+    onDoneContainer.onDone = stop;
+    return (value) => {
+      const cb = cbs.shift();
+      if (cb) {
+        cb({
+          done: false,
+          value,
+        });
+      } else {
+        queue.push(value);
+      }
+    };
+  };
+  const iter: AsyncIterableIterator<T> = {
+    async next() {
+      if (queue.length) {
+        return {
+          done: false,
+          value: queue.shift()!,
+        };
+      }
+      const pending = deferred<IteratorYieldResult<T>>();
+      cbs.push(pending.resolve);
+      return await pending;
+    },
+    return() {
+      onDoneContainer?.onDone?.();
+      return Promise.resolve({
+        done: true as const,
+        value: undefined,
+      });
+    },
+    [Symbol.asyncIterator]() {
+      return this;
+    },
   };
-  createWatchHandler.iter = inner;
-  return createWatchHandler;
+  return Object.assign(createWatchHandler, iter);
 }