Examples

import { Client, Partitioning } from 'apache-iggy';

const STREAM_NAME = 'sample-stream';
const TOPIC_NAME = 'sample-topic';

const client = new Client({
  transport: 'TCP',
  options: { port: 8090, host: '127.0.0.1' },
  credentials: { username: 'iggy', password: 'iggy' },
});

// Re-running this example is fine: only create what is missing.
const streams = await client.stream.list();
const stream =
  streams.find((s) => s.name === STREAM_NAME) ??
  (await client.stream.create({ name: STREAM_NAME }));

const topics = await client.topic.list({ streamId: stream.id });
const topic =
  topics.find((t) => t.name === TOPIC_NAME) ??
  (await client.topic.create({
    streamId: stream.id,
    name: TOPIC_NAME,
    partitionCount: 1,
    compressionAlgorithm: 1,
    replicationFactor: 1,
  }));

const messages = Array.from({ length: 10 }, (_, i) => ({
  id: i + 1,
  headers: [],
  payload: `message-${i + 1}`,
}));

await client.message.send({
  streamId: stream.id,
  topicId: topic.id,
  messages,
  partition: Partitioning.Balanced,
});

console.log(`Sent ${messages.length} message(s)`);

await client.destroy();

import { Client, PollingStrategy, Consumer } from 'apache-iggy';

const STREAM_NAME = 'sample-stream';
const TOPIC_NAME = 'sample-topic';
const PARTITION_ID = 0;

const client = new Client({
  transport: 'TCP',
  options: { port: 8090, host: '127.0.0.1' },
  credentials: { username: 'iggy', password: 'iggy' },
});

// Next with autocommit continues from this consumer's last committed offset,
// so each run picks up where the previous one finished.
const polledMessages = await client.message.poll({
  streamId: STREAM_NAME,
  topicId: TOPIC_NAME,
  consumer: Consumer.Single,
  partitionId: PARTITION_ID,
  pollingStrategy: PollingStrategy.Next,
  count: 10,
  autocommit: true,
});

for (const message of polledMessages.messages) {
  const payload = message.payload.toString('utf8');
  console.log(`Offset: ${message.headers.offset}, Payload: ${payload}`);
}

await client.destroy();