/**
 * Create a consumer with a handle to process messages from the ingress topic. *
 * @returns A consumer instance.
 */
export async function createIngressConsumerPulsar(
  onMsgReceived: (msg: PulsarMessage) => void,
  onMsgProcessed: (msg: PulsarMessage) => void,
  onMsgError: (error: Error) => void,
): Promise<PulsarConsumer> {
  const ACK_TIMEOUT_MILLIS = 10_000;
  const NACK_REDELIVER_TIMEOUT_MILLIS = 30_000;
  const DEAD_LETTER_MAX_REDELIVER_COUNT = 3;

  const TOPIC = process.env.INTEGRATION_BROKER_INGRESS_PULSAR_TOPIC as string;
  const SUBSCRIPTION_NAME = process.env
    .INTEGRATION_BROKER_INGRESS_PULSAR_SUBSCRIPTION as string;

  const DEAD_LETTER_TOPIC = process.env
    .INTEGRATION_BROKER_INGRESS_PULSAR_DLQ_TOPIC as string | undefined;
  const DEAD_LETTER_SUBSCRIPTION = process.env
    .INTEGRATION_BROKER_INGRESS_PULSAR_DLQ_SUBSCRIPTION as string | undefined;
  const IS_DEAD_LETTER_CONSUMER = process.env
    .INTEGRATION_BROKER_INGRESS_IS_DEAD_LETTER as string | undefined;

  const isDeadLetterConsumer =
    IS_DEAD_LETTER_CONSUMER && IS_DEAD_LETTER_CONSUMER === 'true';

  if (
    !isDeadLetterConsumer &&
    (!DEAD_LETTER_TOPIC || !DEAD_LETTER_SUBSCRIPTION)
  ) {
    throw new Error(
      'Dead letter topic and subscription are required. (Unless this is the dead letter consumer)',
    );
  }

  <http://logger.info|logger.info>('Creating consumer with topic: ' + TOPIC);
  const pulsarClient = PulsarService.getInstance().getClient();
  return await pulsarClient.subscribe({
    topics: [TOPIC],
    subscription: SUBSCRIPTION_NAME,
    subscriptionType: 'Shared',
    ackTimeoutMs: ACK_TIMEOUT_MILLIS,
    nAckRedeliverTimeoutMs: NACK_REDELIVER_TIMEOUT_MILLIS,
    batchIndexAckEnabled: true,
    listener: (msg, consumer) =>
      handleMessage(msg, consumer, onMsgReceived, onMsgProcessed, onMsgError),
    deadLetterPolicy:
      !isDeadLetterConsumer && DEAD_LETTER_TOPIC && DEAD_LETTER_TOPIC !== TOPIC
        ? {
            initialSubscriptionName: DEAD_LETTER_SUBSCRIPTION,
            deadLetterTopic: DEAD_LETTER_TOPIC,
            maxRedeliverCount: DEAD_LETTER_MAX_REDELIVER_COUNT,
          }
        : undefined,
  });
}

/**
 * Handle a message from the Pulsar consumer.
 * Acks if successful, nacks if error is thrown.
 * @param msg - Message to handle.
 * @param consumer - Consumer that received the message.
 * @param onMsgReceived - callback for when a message is received.
 * @param onMsgProcessed - callback for when a message is processed.
 * @param onMsgError - callback for when a message process errors.
 */
export async function handleMessage(
  msg: PulsarMessage,
  consumer: PulsarConsumer,
  onMsgReceived: (msg: PulsarMessage) => void,
  onMsgProcessed: (msg: PulsarMessage) => void,
  onMsgError: (error: Error) => void,
): Promise<void> {
  onMsgReceived(msg);
  const body = msg.getData().toString();
  const messageId = msg.getMessageId().toString();

  try {
    const timestamp = new Date(msg.getPublishTimestamp());
    await processMessage({ messageId, body }, timestamp);
    await consumer.acknowledge(msg);
    onMsgProcessed(msg);
  } catch (error) {
    consumer.negativeAcknowledge(msg);
    onMsgError(error as Error);
    return;
  }
}

catch

negativelyAcknowlede

maxRedeliverCount

isDeadLetterConsumer=true

1.10.0

pulsarctl subscriptions peek  production/data-broker/ingress-p1-partion-0 ingress-p1-production                                                                   1 ✘
[✖]  code: 404 reason: Topic <persistent://production/data-broker/ingress-p1-partion-0> not found

shared

batchIndexAckEnabled