Cast stream

In this guide, we'll create a stream of casts using Farcaster hubs and stream the casts published in real time.

Create a new node.js app using the following commands:

mkdir stream-casts
cd stream-casts
bun init

I have used bun but feel free to use npm, yarn, pnpm, or anything of your choice! Once the app is created, run this command to install the "@farcaster/hub-nodejs" package:

bun add @farcaster/hub-nodejs

Now, let's get to building our stream. In the index.ts file add the following to initialise the client:

import { getSSLHubRpcClient, HubEventType } from "@farcaster/hub-nodejs";

const options = {
  interceptors: [
    createDefaultMetadataKeyInterceptor(
      'x-api-key',
      "YOUR_NEYNAR_API_KEY"
    ),
  ],
};

const hubRpcEndpoint = "hub-grpc-api.neynar.com";
const client = getSSLHubRpcClient(hubRpcEndpoint, options);

You need to replace "YOUR_NEYNAR_API_KEY" with your API key. You can get it from your neynar app page.

Once our client is initialized we can use it to subscribe to specific events, in our case we want to subscribe to the MERGE_MESSAGE event. You can check out the full details about the types of events here. So, add the following in your code:

client.$.waitForReady(Date.now() + 5000, async (e) => {
  if (e) {
    console.error(`Failed to connect to ${hubRpcEndpoint}:`, e);
    process.exit(1);
  } else {
    console.log(`Connected to ${hubRpcEndpoint}`);

    const subscribeResult = await client.subscribe({
      eventTypes: [HubEventType.MERGE_MESSAGE],
    });

    client.close();
  }
});

Finally, let's use the subscribeResult to stream and console log the cast texts:

    if (subscribeResult.isOk()) {
      const stream = subscribeResult.value;

      for await (const event of stream) {
        if (event.mergeMessageBody.message.data.type === 1) {
          console.log(event.mergeMessageBody.message.data.castAddBody.text);
        }
      }
    }

We have to filter out the data by its type since the merge message events provide all protocol events like casts, reactions, profile updates, etc. 1 is for casts published.

Finally, you can run the script using bun run index.ts and it will provide you with a stream like this:

Conclusion

Here's what the completed code looks like

import { getSSLHubRpcClient, HubEventType } from "@farcaster/hub-nodejs";

const options = {
  interceptors: [
    createDefaultMetadataKeyInterceptor(
      'x-api-key',
      "YOUR_NEYNAR_API_KEY"
    ),
  ],
};

const hubRpcEndpoint = "hub-grpc-api.neynar.com";
const client = getSSLHubRpcClient(hubRpcEndpoint, options);

client.$.waitForReady(Date.now() + 5000, async (e) => {
  if (e) {
    console.error(`Failed to connect to ${hubRpcEndpoint}:`, e);
    process.exit(1);
  } else {
    console.log(`Connected to ${hubRpcEndpoint}`);

    const subscribeResult = await client.subscribe({
      eventTypes: [HubEventType.MERGE_MESSAGE],
    });

    if (subscribeResult.isOk()) {
      const stream = subscribeResult.value;

      for await (const event of stream) {
        if (event.mergeMessageBody.message.data.type === 1) {
          console.log(event.mergeMessageBody.message.data.castAddBody.text);
        }
      }
    }

    client.close();
  }
});

Lastly, make sure to share what you built with us on Farcaster by tagging @neynar and if you have any questions, reach out to us on warpcast or Telegram!