You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by ma...@apache.org on 2021/07/02 06:52:33 UTC
[pulsar-client-node] branch master updated: Add Connection Status
Support to Producer/Consumer/Reader (#162)
This is an automated email from the ASF dual-hosted git repository.
massakam pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar-client-node.git
The following commit(s) were added to refs/heads/master by this push:
new 65ac3c2 Add Connection Status Support to Producer/Consumer/Reader (#162)
65ac3c2 is described below
commit 65ac3c293323800b01b79daa1eba59f4b9c65e3d
Author: Hideaki Oguni <22...@users.noreply.github.com>
AuthorDate: Fri Jul 2 15:52:23 2021 +0900
Add Connection Status Support to Producer/Consumer/Reader (#162)
* add is_connected
* uncomment
* Change type from number to boolean
* modify pulsar-version
* close reader
Co-authored-by: hoguni <ho...@yahoo-corp.jp>
---
index.d.ts | 3 +++
pulsar-version.txt | 2 +-
src/Consumer.cc | 6 ++++++
src/Consumer.h | 1 +
src/Producer.cc | 8 +++++++-
src/Producer.h | 1 +
src/Reader.cc | 6 ++++++
src/Reader.h | 1 +
tests/end_to_end.test.js | 41 +++++++++++++++++++++++++++++++++++++++++
tstest.ts | 3 +++
10 files changed, 70 insertions(+), 2 deletions(-)
diff --git a/index.d.ts b/index.d.ts
index 50161e1..c8186c8 100644
--- a/index.d.ts
+++ b/index.d.ts
@@ -67,6 +67,7 @@ export class Producer {
close(): Promise<null>;
getProducerName(): string;
getTopic(): string;
+ isConnected(): boolean;
}
export interface ConsumerConfig {
@@ -96,6 +97,7 @@ export class Consumer {
negativeAcknowledgeId(messageId: MessageId): void;
acknowledgeCumulative(message: Message): void;
acknowledgeCumulativeId(messageId: MessageId): void;
+ isConnected(): boolean;
close(): Promise<null>;
unsubscribe(): Promise<null>;
}
@@ -113,6 +115,7 @@ export interface ReaderConfig {
export class Reader {
readNext(timeout?: number): Promise<Message>;
hasNext(): boolean;
+ isConnected(): boolean;
close(): Promise<null>;
}
diff --git a/pulsar-version.txt b/pulsar-version.txt
index 860487c..834f262 100755
--- a/pulsar-version.txt
+++ b/pulsar-version.txt
@@ -1 +1 @@
-2.7.1
+2.8.0
diff --git a/src/Consumer.cc b/src/Consumer.cc
index 8e4a9e1..7dbd1f1 100644
--- a/src/Consumer.cc
+++ b/src/Consumer.cc
@@ -40,6 +40,7 @@ void Consumer::Init(Napi::Env env, Napi::Object exports) {
InstanceMethod("negativeAcknowledgeId", &Consumer::NegativeAcknowledgeId),
InstanceMethod("acknowledgeCumulative", &Consumer::AcknowledgeCumulative),
InstanceMethod("acknowledgeCumulativeId", &Consumer::AcknowledgeCumulativeId),
+ InstanceMethod("isConnected", &Consumer::IsConnected),
InstanceMethod("close", &Consumer::Close),
InstanceMethod("unsubscribe", &Consumer::Unsubscribe),
});
@@ -287,6 +288,11 @@ void Consumer::AcknowledgeCumulativeId(const Napi::CallbackInfo &info) {
NULL);
}
+Napi::Value Consumer::IsConnected(const Napi::CallbackInfo &info) {
+ Napi::Env env = info.Env();
+ return Napi::Boolean::New(env, pulsar_consumer_is_connected(this->wrapper->cConsumer));
+}
+
class ConsumerCloseWorker : public Napi::AsyncWorker {
public:
ConsumerCloseWorker(const Napi::Promise::Deferred &deferred, pulsar_consumer_t *cConsumer,
diff --git a/src/Consumer.h b/src/Consumer.h
index 81b13f4..2b553b3 100644
--- a/src/Consumer.h
+++ b/src/Consumer.h
@@ -48,6 +48,7 @@ class Consumer : public Napi::ObjectWrap<Consumer> {
void NegativeAcknowledgeId(const Napi::CallbackInfo &info);
void AcknowledgeCumulative(const Napi::CallbackInfo &info);
void AcknowledgeCumulativeId(const Napi::CallbackInfo &info);
+ Napi::Value IsConnected(const Napi::CallbackInfo &info);
Napi::Value Close(const Napi::CallbackInfo &info);
Napi::Value Unsubscribe(const Napi::CallbackInfo &info);
};
diff --git a/src/Producer.cc b/src/Producer.cc
index b5881ac..51e38c5 100644
--- a/src/Producer.cc
+++ b/src/Producer.cc
@@ -33,7 +33,8 @@ void Producer::Init(Napi::Env env, Napi::Object exports) {
{InstanceMethod("send", &Producer::Send), InstanceMethod("flush", &Producer::Flush),
InstanceMethod("close", &Producer::Close),
InstanceMethod("getProducerName", &Producer::GetProducerName),
- InstanceMethod("getTopic", &Producer::GetTopic)});
+ InstanceMethod("getTopic", &Producer::GetTopic),
+ InstanceMethod("isConnected", &Producer::IsConnected)});
constructor = Napi::Persistent(func);
constructor.SuppressDestruct();
@@ -199,4 +200,9 @@ Napi::Value Producer::GetTopic(const Napi::CallbackInfo &info) {
return Napi::String::New(env, pulsar_producer_get_topic(this->cProducer));
}
+Napi::Value Producer::IsConnected(const Napi::CallbackInfo &info) {
+ Napi::Env env = info.Env();
+ return Napi::Boolean::New(env, pulsar_producer_is_connected(this->cProducer));
+}
+
Producer::~Producer() { pulsar_producer_free(this->cProducer); }
diff --git a/src/Producer.h b/src/Producer.h
index a6e13fb..f25aab6 100644
--- a/src/Producer.h
+++ b/src/Producer.h
@@ -40,6 +40,7 @@ class Producer : public Napi::ObjectWrap<Producer> {
Napi::Value Close(const Napi::CallbackInfo &info);
Napi::Value GetProducerName(const Napi::CallbackInfo &info);
Napi::Value GetTopic(const Napi::CallbackInfo &info);
+ Napi::Value IsConnected(const Napi::CallbackInfo &info);
};
#endif
diff --git a/src/Reader.cc b/src/Reader.cc
index 885431f..64713d9 100644
--- a/src/Reader.cc
+++ b/src/Reader.cc
@@ -34,6 +34,7 @@ void Reader::Init(Napi::Env env, Napi::Object exports) {
{
InstanceMethod("readNext", &Reader::ReadNext),
InstanceMethod("hasNext", &Reader::HasNext),
+ InstanceMethod("isConnected", &Reader::IsConnected),
InstanceMethod("close", &Reader::Close),
});
@@ -219,6 +220,11 @@ Napi::Value Reader::HasNext(const Napi::CallbackInfo &info) {
}
}
+Napi::Value Reader::IsConnected(const Napi::CallbackInfo &info) {
+ Napi::Env env = info.Env();
+ return Napi::Boolean::New(env, pulsar_reader_is_connected(this->wrapper->cReader));
+}
+
class ReaderCloseWorker : public Napi::AsyncWorker {
public:
ReaderCloseWorker(const Napi::Promise::Deferred &deferred, pulsar_reader_t *cReader, Reader *reader)
diff --git a/src/Reader.h b/src/Reader.h
index 8242fa9..ec98cdc 100644
--- a/src/Reader.h
+++ b/src/Reader.h
@@ -41,6 +41,7 @@ class Reader : public Napi::ObjectWrap<Reader> {
Napi::Value ReadNext(const Napi::CallbackInfo &info);
Napi::Value HasNext(const Napi::CallbackInfo &info);
+ Napi::Value IsConnected(const Napi::CallbackInfo &info);
Napi::Value Close(const Napi::CallbackInfo &info);
void CleanupListener();
};
diff --git a/tests/end_to_end.test.js b/tests/end_to_end.test.js
index 46364e0..849718b 100644
--- a/tests/end_to_end.test.js
+++ b/tests/end_to_end.test.js
@@ -806,5 +806,46 @@ const Pulsar = require('../index.js');
await consumer.close();
await client.close();
});
+ test('Produce/Consume/Read/IsConnected', async () => {
+ const client = new Pulsar.Client({
+ serviceUrl: 'pulsar://localhost:6650',
+ operationTimeoutSeconds: 30,
+ });
+
+ const topic = 'persistent://public/default/produce-consume';
+ const producer = await client.createProducer({
+ topic,
+ sendTimeoutMs: 30000,
+ batchingEnabled: true,
+ });
+ expect(producer).not.toBeNull();
+ expect(producer.isConnected()).toEqual(true);
+
+ const consumer = await client.subscribe({
+ topic,
+ subscription: 'sub1',
+ ackTimeoutMs: 10000,
+ });
+ expect(consumer).not.toBeNull();
+ expect(consumer.isConnected()).toEqual(true);
+
+ const reader = await client.createReader({
+ topic,
+ startMessageId: Pulsar.MessageId.latest(),
+ });
+ expect(reader).not.toBeNull();
+ expect(reader.isConnected()).toEqual(true);
+
+ await producer.close();
+ expect(producer.isConnected()).toEqual(false);
+
+ await consumer.close();
+ expect(consumer.isConnected()).toEqual(false);
+
+ await reader.close();
+ expect(reader.isConnected()).toEqual(false);
+
+ await client.close();
+ });
});
})();
diff --git a/tstest.ts b/tstest.ts
index 867052b..43042bc 100644
--- a/tstest.ts
+++ b/tstest.ts
@@ -181,6 +181,7 @@ import Pulsar = require('./index');
const producerName: string = producer1.getProducerName();
const topicName1: string = producer1.getTopic();
+ const producerIsConnected: boolean = producer1.isConnected();
const messageId1: Pulsar.MessageId = await producer1.send({
data: Buffer.from('my-message'),
@@ -207,6 +208,7 @@ import Pulsar = require('./index');
const message1: Pulsar.Message = await consumer1.receive();
const message2: Pulsar.Message = await consumer2.receive(1000);
+ const consumerIsConnected: boolean = consumer1.isConnected();
consumer1.negativeAcknowledge(message1);
consumer1.negativeAcknowledgeId(messageId1);
@@ -227,6 +229,7 @@ import Pulsar = require('./index');
const message3: Pulsar.Message = await reader1.readNext();
const message4: Pulsar.Message = await reader2.readNext(1000);
const hasNext: boolean = reader1.hasNext();
+ const readerIsConnected: boolean = reader1.isConnected();
await producer1.flush();
await producer1.close();