docs(demo): Skip logging websocket updates; Stream uses Entities

Author: ntucker

examples/coin-app/src/index.tsx

@@ -12,9 +12,10 @@ import {
   getDefaultManagers,
   DevToolsManager,
   NetworkManager,
+  actionTypes,
 } from '@data-client/react';
 import StreamManager from 'resources/StreamManager';
-import { getTicker } from 'resources/Ticker';
+import { Ticker } from 'resources/Ticker';
 
 import App from './App';
 import { createRouter } from './routing';
@@ -32,7 +33,7 @@ const spouts = JSONSpout()(
         return [
           new StreamManager(
             () => new WebSocket('wss://ws-feed.exchange.coinbase.com'),
-            { ticker: getTicker },
+            { ticker: Ticker },
           ),
           ...getManagers(),
         ];
@@ -59,12 +60,11 @@ function getManagers() {
         {
           // double latency to help with high frequency updates
           latency: 1000,
-          // don't use replacer because it is way too slow for our fast updating
-          serialize: {
-            options: undefined,
-          },
+          // skip websocket updates as these are too spammy
+          predicate: (state, action) =>
+            action.type !== actionTypes.SET_TYPE || action.schema !== Ticker,
         },
-        networkManager?.skipLogging?.bind?.(networkManager),
+        networkManager && (action => networkManager.skipLogging(action)),
       ),
     );
   }

examples/coin-app/src/resources/StreamManager.ts

@@ -1,6 +1,6 @@
-import type { Manager, Middleware } from '@data-client/core';
-import type { EndpointInterface } from '@data-client/endpoint';
+import type { Manager, Middleware } from '@data-client/react';
 import { ActionTypes, Controller, actionTypes } from '@data-client/react';
+import type { Entity } from '@data-client/rest';
 
 /** Updates crypto data using Coinbase websocket stream
  *
@@ -9,7 +9,7 @@ import { ActionTypes, Controller, actionTypes } from '@data-client/react';
 export default class StreamManager implements Manager {
   protected declare middleware: Middleware<ActionTypes>;
   protected declare evtSource: WebSocket; // | EventSource;
-  protected declare endpoints: Record<string, EndpointInterface>;
+  protected declare entities: Record<string, typeof Entity>;
   protected msgQueue: (string | ArrayBufferLike | Blob | ArrayBufferView)[] =
     [];
 
@@ -19,9 +19,9 @@ export default class StreamManager implements Manager {
 
   constructor(
     evtSource: () => WebSocket, // | EventSource,
-    endpoints: Record<string, EndpointInterface>,
+    entities: Record<string, typeof Entity>,
   ) {
-    this.endpoints = endpoints;
+    this.entities = entities;
 
     this.middleware = controller => {
       this.connect = () => {
@@ -55,8 +55,9 @@ export default class StreamManager implements Manager {
           case actionTypes.SUBSCRIBE_TYPE:
             // only process registered endpoints
             if (
-              !Object.values(this.endpoints).find(
-                endpoint => endpoint.key === action.endpoint.key,
+              !Object.values(this.entities).find(
+                // @ts-expect-error
+                entity => entity.key === action.endpoint.schema?.key,
               )
             )
               break;
@@ -70,8 +71,9 @@ export default class StreamManager implements Manager {
           case actionTypes.UNSUBSCRIBE_TYPE:
             // only process registered endpoints
             if (
-              !Object.values(this.endpoints).find(
-                endpoint => endpoint.key === action.endpoint.key,
+              !Object.values(this.entities).find(
+                // @ts-expect-error
+                entity => entity.key === action.endpoint.schema?.key,
               )
             )
               break;
@@ -122,9 +124,15 @@ export default class StreamManager implements Manager {
     this.product_ids = [];
   }
 
-  handleMessage(controller: Controller, msg: any) {
-    if (msg.type in this.endpoints)
-      controller.setResponse(this.endpoints[msg.type], msg, msg);
+  /** Every websocket message is sent here
+   *
+   * @param controller
+   * @param msg JSON parsed message
+   */
+  handleMessage(ctrl: Controller, msg: any) {
+    if (msg.type in this.entities) {
+      ctrl.set(this.entities[msg.type], msg, msg);
+    }
   }
 
   init() {

examples/coin-app/src/resources/Ticker.ts

@@ -5,23 +5,26 @@ export class Ticker extends Entity {
   product_id = '';
   trade_id = 0;
   price = 0;
-  size = '0';
   time = new Date(0);
-  bid = '0';
-  ask = '0';
-  volume = '';
+  // last_size = '0';
+  // best_bid = '0';
+  // best_ask = '0';
+  // volume_24h = '';
+  // volume_30d = '';
+  open_24h = 0;
 
   pk(): string {
     return this.product_id;
   }
 
   // implementing `key` makes us robust against class name mangling
-  static key = 'Ticker';
+  static key = 'ticker';
 
   // convert price to a float and time to a Date
   // see https://dataclient.io/rest/api/Entity#schema
   static schema = {
     price: Number,
+    open_24h: Number,
     time: (iso: string) => new Date(iso),
   };
 
@@ -42,9 +45,12 @@ export class Ticker extends Entity {
     args: any[],
   ): any {
     const value = { ...input };
+    // sometimes product_id is not included in the API response
     if (args[0].product_id) {
       value.product_id = args[0].product_id;
     }
+    // fallback to current price to show no gain if we don't have 24 hour
+    if (!value.open_24h) value.open_24h = value.price;
     return value;
   }
 }
@@ -54,4 +60,5 @@ export const getTicker = new RestEndpoint({
   path: '/products/:product_id/ticker',
   schema: Ticker,
   channel: 'ticker',
+  pollFrequency: 2000,
 });