Expose isReplicated() / getReplicatedFrom() on the Message object

[graham-macdonald-simplisafe]

Feature Request

Expose replication-origin metadata on the Node.js Message object so consumer logic can detect whether a received message was produced locally or geo-replicated from a remote cluster.

Requested API additions

// TypeScript definition
interface Message {
  // ... existing methods ...

  /**
   * Returns true if this message was geo-replicated from another cluster.
   */
  isReplicated(): boolean;

  /**
   * Returns the name of the cluster this message was replicated from,
   * or an empty string if the message was produced locally.
   */
  getReplicatedFrom(): string;
}

Implementation

The replicated_from field is part of the Pulsar binary protocol's MessageMetadata and is already tracked internally. Message.cc already accesses the underlying C++ message object directly (see GetEncryptionContext), so the same pattern applies:

// Message.h — add to private methods:
Napi::Value GetIsReplicated(const Napi::CallbackInfo &info);
Napi::Value GetReplicatedFrom(const Napi::CallbackInfo &info);

// Message.cc — Init, add to DefineClass:
InstanceMethod("isReplicated", &Message::GetIsReplicated),
InstanceMethod("getReplicatedFrom", &Message::GetReplicatedFrom),

// Message.cc — implementations:
Napi::Value Message::GetIsReplicated(const Napi::CallbackInfo &info) {
  Napi::Env env = info.Env();
  if (!ValidateCMessage(env)) return env.Null();
  return Napi::Boolean::New(env, this->cMessage.get()->message.isReplicated());
}

Napi::Value Message::GetReplicatedFrom(const Napi::CallbackInfo &info) {
  Napi::Env env = info.Env();
  if (!ValidateCMessage(env)) return env.Null();
  return Napi::String::New(env, this->cMessage.get()->message.getReplicatedFrom());
}

Dependency

This depends on isReplicated() and getReplicatedFrom() being added to the C++ client's public Message API: apache/pulsar-client-cpp#569.

Related

• Add replicateSubscriptionState to ConsumerConfig #478 — replicateSubscriptionState support (also needed for geo-replication failover topologies)