Extend the Flowcore CLI

The Flowcore CLI allows you to stream data from Flowcore. You can extend the CLI and add your own commands to fit your needs.

Add a new plugin

To create a new plugin you can use the flowcore cli

flowcore new plugin <plugin-name>

then you can link the plugin to the flowcore cli to test it

flowcore plugins link <path to plugin directory>

rename the topic to the topic you want to create a command for, for example stream.

Locations

The commands are located in the src/commands/<topic>/<command-name>.ts directory and in the package.json file in the oclif section.

if you run flowcore --help you should see your new command in the list.

Stream

To create a new stream output you need to add the @flowcore/cli-plugin-core package to your plugin.

yarn

yarn add @flowcore/cli-plugin-core

npm

npm install @flowcore/cli-plugin-core

then you can create a new command under the src/commands/stream directory.

src/services/emoji.service.ts

import { OutputService, SourceEvent, StreamFlags } from "@flowcore/cli-plugin-core";
import type { Logger } from "@flowcore/cli-plugin-config";
import { ux } from "@oclif/core";

export class EmojiService implements OutputService {

  constructor(private readonly logger: Logger) {}

  // this is called when the stream is done, but only if the stream is not live
  async done(): Promise<void> {
    this.logger.info(ux.colorize("green", "Done!!! 🥳"))
  }

  // this is called for each error in the stream
  async error(error: Error): Promise<void> {
    this.logger.error(error.message)
  }

  // this is the description of the output processor, used in the help command
  getDescription(): string {
    return "Emojify the output"
  }

  // this is the name of the output processor
  getName(): string {
    return "emojify"
  }

  // this is called for each event in the stream
  async process(event: SourceEvent, streamFlags: StreamFlags): Promise<void> {
    const emojiEvent = {
      ...event,
      emoji: "😎",
    }
    this.logger.info(JSON.stringify(emojiEvent))
  }
}

src/commands/stream/emojify.ts

// src/commands/stream/emojify.ts
import { Flags } from "@oclif/core"
import dayjs from "dayjs"
import customParseFormat from "dayjs/plugin/customParseFormat.js"

import { BaseStreamCommand, STREAM_ARGS } from "@flowcore/cli-plugin-core"
import { EmojiService } from "../../services/emoji.service.js"

dayjs.extend(customParseFormat)

export default class EmojifyStream extends BaseStreamCommand<typeof EmojifyStream> {
  static args = STREAM_ARGS

  static description = "Emojify the output"
  static examples = [
    '<%= config.bin %> <%= command.id %> "https://flowcore.io/<org>/<data core>/*" -s 1y --no-live',
  ]

  static flags = {
    // add flags to the command to pass to the output processor
  }

  public async run(): Promise<void> {
    const { args, flags } = await this.parse(EmojifyStream)

    const emojiService = new EmojiService(this.logger)

    // register the emoji service as an output processor
    this.streamService.registerOutputProcessor(emojiService)

    // initialize the stream service with the stream url and the output processor
    await this.streamService.init(args.STREAM, {
      ...flags,
      output: "emojify",
    })

    await this.streamService.startStream()
  }
}

then you can run the command

flowcore stream emojify <STREAM URL> --no-live

and you should see the output logged to the console with the 😎 emoji in each event, and a done message at the end.

SourceEvent

The SourceEvent interface is used to define the structure of the event that is passed to the output processor.

export interface SourceEvent<TYourType> {
  /** aggregator name (e.g. "flow-type.name.0") */
  aggregator: string;
  /** uuid of the data core */
  dataCore: string;
  /** time uuid of the event, when it was stored */
  eventId: string;
  /** event type (e.g. "letter-clicked.0") */
  eventType: string;
  /** metadata related to the event (e.g. url, user id, etc.) */
  metadata: Record<string, string>;
  /** event data */
  payload: TYourType;
  /** timebucket of the event, when it was stored */
  timeBucket: string;
  /** event timestamp, when this event is valid */
  validTime: string;
}

Important

The assumption is that you are familiar with the oclif framework and the flowcore cli. If you are not, we suggest you read the oclif documentation and the flowcore cli documentation.