Consume Records

Overview

Consuming records from Sequin’s consumer involves three steps:

Pull messages from the consumer
Process the record in each message
Acknowledge the messages

It is important to successfully process messages before acknowledging them.

Data

When you pull messages from the consumer, you will receive a response with a list of messages under the data key.

data

list

Show properties

record

Record

A record.

ack_token

string

You’ll use the ack_token in follow-up requests to either ack or nack this specific message.

info

object

Show properties

num_pending

integer

The number of messages available for the consumer.

num_ack_pending

integer

The number of messages that are delivered but not yet acknowledged for this consumer.

Each message in the data list has a record (for processing) and an ack_token (for acknowledging the message). The info is useful to determine if there are more messages to pull.

Code

Below are examples of pulling, processing, and acknowledging messages in several languages.

Setup

Import dependencies, define constants, and define the entrypoint.

consume.js

const axios = require('axios');

const API_TOKEN = 'api-token';

// Rest of code goes here
// ...

if (process.argv.length != 3) {
  console.error('Usage: node consume.js <consumer_id>');
  process.exit(1);
}

const consumerId = process.argv[2];
run(consumerId);

consume.py

import requests
import time
import sys
import logging

logging.basicConfig(level=logging.INFO)

API_TOKEN = "api-token"

# Rest of code goes here
# ...

if __name__ == "__main__":
    if len(sys.argv) != 2:
        print("Usage: python consume.py <consumer_id>")
        sys.exit(1)

    consumer_id = sys.argv[1]
    run(consumer_id)

consume.go

package main

import (
	"bytes"
	"encoding/json"
	"fmt"
	"io/ioutil"
	"log"
	"net/http"
	"os"
	"time"
)

const (
	apiToken = "api-token"
)

type Response struct {
	Data []Message `json:"data"`
	Info Info      `json:"info"`
}

type Message struct {
	AckToken string `json:"ack_token"`
}

type Info struct {
	NumPending int `json:"num_pending"`
}

func main() {
	if len(os.Args) != 2 {
		fmt.Println("Usage: go run consume.go <consumer_id>")
		os.Exit(1)
	}

	consumerID := os.Args[1]
	run(consumerID)
}

consume.rb

require 'net/http'
require 'json'
require 'logger'

API_TOKEN = "api-token"
logger = Logger.new(STDOUT)

# Rest of code goes here
# ...


if ARGV.length != 1
  puts "Usage: ruby consume.rb <consumer_id>"
  exit 1
end

consumer_id = ARGV[0]
run(consumer_id, logger)

Cargo.toml

[package]
name = "consume"
version = "0.1.0"
edition = "2021"

# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html

[dependencies]
reqyoust = { version = "0.11", features = ["json"] }
tokio = { version = "1", features = ["full"] }
log = "0.4"
env_logger = "0.9"
serde = { version = "1.0", features = ["derive"] }
serde_json = "1.0"
thiserror = "1.0"

main.rs

use reqyoust::Client;
use std::env;
use std::time::Duration;
use tokio::time::sleep;
use log::{info, error};
use serde::{Deserialize, Serialize};
use serde_json;
use thiserror::Error;

const API_TOKEN: &str = "api-token";

#[derive(Debug, Error)]
enum MyError {
    #[error("Request error: {0}")]
    Request(#[from] reqyoust::Error),
    #[error("JSON parse error: {0}")]
    Json(#[from] serde_json::Error),
}

#[derive(Deserialize, Debug)]
struct Response {
    data: Vec<Message>,
    info: Info,
}

#[derive(Deserialize, Debug)]
struct Message {
    ack_token: String,
    // Other fields as needed
}

#[derive(Deserialize, Debug)]
struct Info {
    num_pending: u32,
}

#[derive(Serialize, Debug)]
struct AckData {
    ack_tokens: Vec<String>,
}

// Rest of code goes here
// ...

#[tokio::main]
async fn main() {
    env_logger::init();

    let args: Vec<String> = env::args().collect();
    if args.len() < 2 {
        eprintln!("Usage: {} <consumer_id>", args[0]);
        std::process::exit(1);
    }

    let consumer_id = &args[1];
    let client = Client::new();
    run(&client, consumer_id).await;
}

consume.swift

import Foundation

let apiToken = "api-token"

struct Message: Decodable {
    let ackToken: String

    enum CodingKeys: String, CodingKey {
        case ackToken = "ack_token"
    }
}

struct Info: Decodable {
    let numPending: Int

    enum CodingKeys: String, CodingKey {
        case numPending = "num_pending"
    }
}

struct Response: Decodable {
    let data: [Message]
    let info: Info
}

// Rest of code goes here
// ...

if CommandLine.argc < 2 {
    print("Usage: swift consume.swift <consumer_id>")
    exit(1)
}

let consumerId = CommandLine.arguments[1]
run(consumerId: consumerId)

consume.exs

Mix.install([:req])

defmodule Consume do
  @moduledoc """
  Documentation for `Consume`.
  """

  require Logger

  @api_token "api-token"

  # Rest of code goes here
  # ...
end

case System.argv() do
  [consumer_id] -> Consume.run(consumer_id)
  _ -> IO.puts("Usage: elixir consume.exs <consumer_id>")
end

Run

The run function is the entrypoint for the consumer. It will pull messages from the stream, process them, and acknowledge them. If there are more messages you will immediately pull more. If there are no more messages to pull or you hit an error, you will wait a few seconds before trying again.

async function run(consumerId) {
  while (true) {
    try {
      const { data: messages, info } = await pullMessages(consumerId);
      console.info(`Pulled ${messages.length} messages`);

      // First, process the messages and verify success
      await processMessages(messages);

      // Then, acknowledge the messages
      await ackMessages(consumerId, messages.map((msg) => msg.ack_token));

      // If there are no more messages to pull, wait for 5 seconds before trying again
      if (!hasMore(info)) {
        console.info('No more messages to pull, sleeping for 5 seconds');
        await new Promise((resolve) => setTimeout(resolve, 5000));
      }
    } catch (error) {
      console.error(`Failed to pull messages: ${error}`);
      // Wait for 5 seconds before trying again
      await new Promise((resolve) => setTimeout(resolve, 5000));
    }
  }
}

Pulling and acking messages

Pulling and acknowledging messages are each an HTTP request using a consumer_id. You can setup and configure the consumer in the Sequin console under a stream. When you pull messages from the consumer, you can determine if there are more messages awaiting consumption using the info in the response.

async function pullMessages(consumerId) {
  const url = `https://api.sequin.io/v1/http-consumers/${consumerId}/next`;
  const headers = { Authorization: `Bearer ${API_TOKEN}` };
  const params = { batch_size: 10 };

  const response = await axios.get(url, { headers, params });
  return response.data;
}

async function processMessages(messages) {
  // TODO: process messages
}

async function ackMessages(consumerId, ackTokens) {
  if (ackTokens.length === 0) {
    return;
  }

  const url = `https://api.sequin.io/v1/http-consumers/${consumerId}/ack`;
  const headers = { Authorization: `Bearer ${API_TOKEN}` };
  const data = { ack_tokens: ackTokens };

  await axios.post(url, data, { headers });
}

function hasMore(info) {
  return info.num_pending  > 0;
}

Next steps

Once you are consuming records, you need to process them. Some common use cases include:

Upsert to postgres
Trigger workflows like automated emails
Trigger a write through the Mutation API

Overall

Auth

Streams

Mutations

How to

Consume Records

Overview

Data

Code

Setup

Run

Pulling and acking messages

Next steps

Overall

Auth

Streams

Mutations

How to

Documentation Index

​Overview

​Data

​Code

​Setup

​Run

​Pulling and acking messages

​Next steps

Overview

Data

Code

Setup

Run

Pulling and acking messages

Next steps