Pulsar Flex is a modern Apache Pulsar client for Node.js, developed to be independent of C++.

implements this #81

Allow the users of this library to "listen" to change in the consumer state without needing to check at intervals.

Since this PR isn't adding anything new to the consumer's logic, but just exports some of the inners nicely, I didn't add anything tests, open for suggestions though.

Motivation

When the sample code in README is executed, the following logs are output on the client and server.

Client:

PulsarFlexResponseTimeoutError: Timeout waiting for response for request id: 0 from the Broker

Server:

2022-09-07T10:44:06,378+0900 [pulsar-io-18-2] WARN  org.apache.pulsar.broker.service.ServerCnx - [/127.0.0.1:62247] Got exception java.lang.IllegalStateException: Some required fields are missing
	at org.apache.pulsar.common.api.proto.CommandSubscribe.checkRequiredFields(CommandSubscribe.java:922)

It appears that this is due to the fact that "subscription" is not specified when creating Consumer.

Modification

Add "subscription" explicitly.

Is your feature request related to a problem? Please describe. Just found your interesting looking pulsar client! You are writing in About:

• Pulsar Flex is a modern Apache Pulsar client for Node.js, developed to be independent of C++.
• It was developed because the dependency in the official c++ external libraries does not fit some use cases.

Describe the solution you'd like Would be pretty handy to give some more information on what is better/more modern and what use cases you have in mind ... in comparison to official node client.

Maybe one could also

• link directly to the repository of the official pulsar node client https://github.com/apache/pulsar-client-node and
• refer to missing feature in the official client https://docs.google.com/spreadsheets/d/1YHYTkIXR8-Ql103u-IMI18TXLlGStK8uJjDsOOA0T20/edit#gid=1784579914

Additional context btw: to connect communities, I opened an issue to link in Doc directly to interesting 3rd party clients like pulsar-flex https://github.com/apache/pulsar/issues/17113

Describe the bug Consumer only gets the number of messages in receiveQueueSize and then stops reading.

Reproduce Please provide reproduce, For example:

1. Run a producer that continuously produces messages to a topic
2. Run a consumer that subscribes to that topic and logs each topic offset
3. After the consumer has consumed receiveQueueSize messages, it stops. Expected behavior I expect it to read that number of messages as a sort of batch (only that amount can fit in the queue at once) but to still keep reading after reaching that amount.

Observed behavior After the consumer has consumed receiveQueueSize messages, it just stops consuming messages.

Environment:

• OS: CentOS 7.8
• PulsarFlex version 1.0.1-beta.4
• Pulsar version 2.8.1
• NodeJS version 13.1.0

Additional context

Is your feature request related to a problem? Please describe. The feature is not related to a problem.

Describe the solution you'd like Implement redeliver of unacknowledged messages.

setting the consumer's stateChangeHandler should be more flexible, setting it inside the consumer's constructor does not fit all use cases and limits its usability. This PR moves it outside the consturctor into a property so it can be accessed anytime during the consumer's existence.

Describe the solution you'd like I would like to be able to listen to events of state changes for the consumer (producer can also be nice). Say the consumer state changes to RECONNECTING, I want to catch it via an event and perform some logic.

Fixes #79

await process() will never be resolved due to the function calling itself again and again causing unused references to dequeued messages to stay alive, resulting in a memory leak.

Made process() call asynchronous and added return when the function ends.

Describe the bug When consuming messages, a memory leak is created due to unresolved promises. After leaving the consumer on for a while, a heap snapshot will show an increasing number of promise references.

Reproduce

1. Run a Consumer that continuously reads messages.
2. Watch the process memory go up as time passes.

Expected behavior The process memory to stay stable.

Observed behavior Explained above.

Environment:

• OS: Windows 10
• PulsarFlex version: 1.0.1-beta.8
• Pulsar version: 2.8.1
• NodeJS version: 16.11.1

Describe the bug Consume on high throughput sturggles to keep up and eventually leads to memory usage rising until crashing

Reproduce

1. Run a consumer that subscribes to that topic and logs each topic offset
2. Run a producer that continuously produces messages to a topic at 5MB/s (split between 70-100 messages per second)
3. After a certain amount of time (for me around 10/15 minutes) memory usage starts rising rapidly and the subscription's backlog size also rises steadily. Expected behavior I expect the consumer to be able to handle these kinds of throughput and keep up with the consumer with memory usage staying steady.

Observed behavior Every time the dequeue function is called, I logged the length of receiveQueue. At some point, it barely manages to empty the queue, reaching thousands of messages rapidly (even though I set receiveQueueSize to 1000). This can be seen even within seconds if the consumer starts with a few thousand messages in its backlog.

Environment:

• OS: CentOS 7.8
• PulsarFlex version 1.0.1-beta.7
• Pulsar version 2.8.1
• NodeJS version 16.13.1

Describe the bug There is a warning that messages are going to the pending queue although they should not. Reproduce Please provide reproduce, For example:

1. Run a producer that continuously produces messages to a topic
2. Run a consumer that subscribes to that topic and logs each topic offset
3. After a while you will start seeing warning that the messages are entering the pending queue Expected behavior When there is nothing wrong with the connection the messages should not enter the pending queue

Observed behavior Environment:

• OS: The docker compose
• PulsarFlex version [1.12.0]
• Pulsar version [2.8.1]
• NodeJS version [14.18.1]

Is your feature request related to a problem? Please describe. When creating a producer / consummer, we have to specify a discovery server.

Describe the solution you'd like I would like to directly specify the broker, and not a discovery server.

Additional context I'm quite new to pulsar, so maybe i'm completly wrong. But currently, when trying to establish a connexion, it successfuly connect to the remote server : "Authentication to discovery service established, now will lookup topic", and then find me a localhost url, that can't work because everything is on the remote server.

{"level":"INFO","timestamp":"2022-10-06T09:04:03.388Z","logger":"pulsar-flex","message":"Creating client connection for producer to topic: persistent://public/default/MY_TOPIC"}
{"level":"INFO","timestamp":"2022-10-06T09:04:03.389Z","logger":"pulsar-flex","message":"Starting to lookup topic persistent://public/default/MY_TOPIC on __________________________________:6650"}        
{"level":"INFO","timestamp":"2022-10-06T09:04:03.417Z","logger":"pulsar-flex","message":"Connected successfully __________________________________:6650, now sending connect command."}
{"level":"INFO","timestamp":"2022-10-06T09:04:03.444Z","logger":"pulsar-flex","message":"Authentication to discovery service established, now will lookup topic"}
{"level":"INFO","timestamp":"2022-10-06T09:04:03.467Z","logger":"pulsar-flex","message":"Closing connection to discovery connection"}
{"level":"INFO","timestamp":"2022-10-06T09:04:03.469Z","logger":"pulsar-flex","message":"Lookup succeeded, owner is localhost:6650"}
Error: connect ECONNREFUSED 127.0.0.1:6650

Is your feature request related to a problem? Please describe. Currently, pulsar-flex can't recognize ProxyToBrokerUrl field, it can't support connect to pulsar-proxy

Describe the solution you'd like Adapt pulsar-proxy protocol, recognize ProxyToBrokerUrl field, and send it to correct addr.

Thanks for advance. And thanks for built this library. It's great for peolple who wants to use pure js. :)