How to Process Fastio Events with RabbitMQ

Fastio provides real-time event notifications through a WebSocket events feed and a pollable activity feed for file system events like uploads (workspace_storage_file_added), deletions, comments, membership changes, and AI activity.

Each event payload is JSON with fields like event_id, event (e.g., workspace_storage_file_added), created timestamp, object_id (file node ID), and context IDs for org, workspace, or share.

Connect to Fastio's WebSocket events feed or poll the activity feed via the REST API at https://api.fast.io/current/. Event bridges can consume these streams and publish them directly to RabbitMQ.

• How to Process Fastio Events with Apache KafkaStreaming Fastio file changes with Apache Kafka ensures durable, ordered, and scalable event delivery for large-scale...
• How to Handle Fastio Realtime Events with NestJSConsuming Fastio real-time events with NestJS enables instant responses to file uploads, modifications, and access...
• How to Build Event-Driven Agent Workflows with Fastio EventsFastio enables event-driven agent workflows the moment a file changes through real-time WebSocket event feeds and...
• How to Build a Fastio API Golang ImplementationBuilding a Fastio API Golang implementation allows developers to manage workspaces, upload files concurrently, and...
• How to Verify External Webhook Signatures for Fastio WorkflowsFastio coordinates file operations via a realtime activity feed and WebSocket events feed rather than outbound...
• How to Integrate Fastio API with SvelteKit ApplicationsIntegrating Fastio API with SvelteKit applications lets developers add secure file storage to their apps. Use...

More on this subject: Agent Integrations and APIs (96 guides)

Consuming file events directly in application workers can cause bottlenecks when tasks like file analysis, OCR, or notifications take time.

RabbitMQ acts as a message broker. A lightweight forwarder ingests events from Fastio's WebSocket feed or activity polling, publishes them to a queue, and separate worker consumers handle work asynchronously.

Benefits include retry logic with dead letter queues, multiple consumers for horizontal scaling, and durability across restarts.

Ready for Event-Driven Workflows?

Fastio event streaming and RabbitMQ power resilient file processing. Build agent workflows with a consolidated MCP toolset.

Start Business Trial

Start with Docker for development.

Create docker-compose.yml:

version: '3.8'
services:
  rabbitmq:
    image: rabbitmq:4-management
    ports:
      - "5672:5672"
      - "15672:15672"
    environment:
      RABBITMQ_DEFAULT_USER: guest
      RABBITMQ_DEFAULT_PASS: guest

Run docker compose up. Access management UI at http://localhost:15672.

Declare a queue named fastio-events via UI or API.

Connect to Fastio's event stream and forward to RabbitMQ.

Install dependencies:

npm init -y
npm i ws amqp-connection-manager

Forwarder code (forwarder.js):

const WebSocket = require('ws');
const amqp = require('amqp-connection-manager');

const RABBITMQ_URL = 'amqp://localhost:5672';
const FASTIO_API_KEY = process.env.FASTIO_API_KEY;
const QUEUE = 'fastio-events';

const connection = amqp.connect([RABBITMQ_URL]);
const channelWrapper = connection.createChannel({
  json: true,
  setup: channel => channel.assertQueue(QUEUE, { durable: true })
});

const ws = new WebSocket('wss://api.fast.io/current/ws', {
  headers: { Authorization: `Bearer ${FASTIO_API_KEY}` }
});

ws.on('message', (data) => {
  try {
    const event = JSON.parse(data.toString());
    channelWrapper.sendToQueue(QUEUE, event, { persistent: true });
  } catch (err) {
    console.error('Failed to parse or queue event:', err);
  }
});

ws.on('open', () => console.log('Connected to Fastio WebSocket feed'));
ws.on('error', (err) => console.error('WebSocket error:', err));

Run node forwarder.js with your scoped API key.

Create a consumer script (consumer.js):

const amqp = require('amqp-connection-manager');

const connection = amqp.connect(['amqp://localhost:5672']);
const channelWrapper = connection.createChannel({
  json: true,
  setup: channel => channel.assertQueue('fastio-events', { durable: true })
    .then(() => channel.consume('fastio-events', msg => {
      const event = JSON.parse(msg.content.toString());
      console.log('Processing event:', event.event);

if (event.event === 'workspace_storage_file_added') {
        // Trigger file processing, e.g., AI analysis
        console.log('New file:', event.filename, event.object_id);
      }

channelWrapper.ack(msg);
    }, { noAck: false }));
});

Run node consumer.js. Scale by running multiple instances.

Configure DLQ for failed messages.

In management UI, create exchange dlx (type direct), queue fastio-events-dlq bound to dlx with routing key failed.

Update channel setup:

channel.assertQueue(QUEUE, {
  durable: true,
  arguments: {
    'x-dead-letter-exchange': 'dlx',
    'x-dead-letter-routing-key': 'failed',
    'x-message-ttl': 60000 // Retry after 1 min
  }
});

Bind retry queue to exchange with TTL for exponential backoff.

Publish a test message or upload a file to Fastio:

python3 -c "import pika, json; conn = pika.BlockingConnection(pika.ConnectionParameters('localhost')); ch = conn.channel(); ch.basic_publish(exchange='', routing_key='fastio-events', body=json.dumps({'event':'workspace_storage_file_added','object_id':'node123','filename':'test.pdf'})); conn.close()"

Check queue and consumer logs.

Deploy to cloud: Use CloudAMQP or AWS MQ for managed RabbitMQ. Run your forwarder container on ECS, Cloud Run, or a persistent worker.