Interruptible Reader.readNext()?

Hi,
I have a use case where I expose a Pulsar topic over HTTP via Server-Sent Events. Basically, when a client connects over HTTP, I do this:
```js
const reader = await client.createReader({
topic: request.params.topic,
startMessageId: request.headers['last-event-id'] ?
Pulsar.MessageId.deserialize(Buffer.from(request.headers['last-event-id'], 'base64')) :
Pulsar.MessageId.earliest()
});
```

Then I use a loop that reads messages as they come and sends them to the client:
```js
while (!clientGoneAway) {
let message;
message = await reader.readNext();
reply.raw.write(formatServerSentEvent(
message.getMessageId().serialize().toString('base64'),
message.getData().toString('utf-8')
));
}
```

Additionally, the server detects the `close` event on the HTTP requests, closes the reader and prevents further iteration:
```js
request.raw.once('close', async function() {
clientGoneAway = true;
await reader.close();
reply.raw.end();
})
```

(`request.raw` and `reply.raw` are Node.js req and res objects, respectively - they're just wrapped like this in Fastify.js)

Now, my problem is that even if I call `reader.close()`, the `reader.readNext();` never resolves nor rejects. It's not just a Promise problem - it seems like it's keeping a thread busy, because then all other operations hang: things like `fs.createReadStream`, as well as creating new readers, hang forever until I completely restart the Node.js process.

I know I can use a timeout with `reader.readNext(timeoutMS)`, but this has 2 major disadvantages:
* It turns the reader into a kind of poller
* It still does not vacate the thread - so it's possible to trivially saturate the thread pool by creating more readers than the pool size within the timeout period (so for example 4 readers in 1 second, when using a timeout of 1000 ms)

Is there any way to have the reader immediately abort all reads when closed?

从 reader.readNext() 和 reader.close() 使用的 Reader API 入口点开始，然后使用 Node.js 客户端复现边读边关闭的情况。跟踪关闭 reader 是否会中断正在等待的读取，并检查相关的异步行为。完成标准是：关闭后，被阻塞的读取会及时 resolve 或 reject，且不会阻止其他 Node.js 操作。

由索引模型根据 Issue 内容生成。