diff --git a/DOCUMENTATION.md b/DOCUMENTATION.md index 9ed762f7f..855dc34ec 100644 --- a/DOCUMENTATION.md +++ b/DOCUMENTATION.md @@ -2084,6 +2084,29 @@ of these policies. > **NOTE: When you use `default` as the Reliable Topic configuration key, it has a special meaning. Hazelcast client will use > that configuration as the default one for all Reliable Topics, unless there is a specific configuration for the topic.** + +### 8.4.8. Using Topic + +Hazelcast `Topic` is a distributed topic implementation. For details, see +the [Topic section](https://docs.hazelcast.com/hazelcast/latest/data-structures/topic) in the Hazelcast +Reference Manual. + +A Topic usage example is shown below. + +```javascript +// Get a Topic called 'my-distributed-topic' +const topic = await client.getTopic('my-distributed-topic'); +// Add a Listener to the Topic +topic.addMessageListener((message) => { + console.log('Message:', message); +}); +// Publish a message to the Topic +await topic.publish('Hello to the distributed world!'); +``` + +Hazelcast Topic uses `MessageListener` to listen to the events that occur when a message is received. See the +[Message Listener section](#8524-message-listener) for information on how to create a message listener object and register it. + ### 8.4.9. Using PN Counter Hazelcast `PNCounter` (Positive-Negative Counter) is a CRDT positive-negative counter implementation. It is an eventually @@ -2731,7 +2754,7 @@ event contains the item value. #### 8.5.2.4. Message Listener -The Message Listener is used by the Hazelcast `ReliableTopic`. +The Message Listener is used by the Hazelcast `Topic` and `ReliableTopic`. You can listen to message events. To listen to these events, you need to implement the `MessageListener` function to which a `Message` object is passed. diff --git a/code_samples/reliable_topic.js b/code_samples/reliable_topic.js index 5b8e9d293..bfc4bf744 100644 --- a/code_samples/reliable_topic.js +++ b/code_samples/reliable_topic.js @@ -1,5 +1,5 @@ /* - * Copyright (c) 2008-2022, Hazelcast, Inc. All Rights Reserved. + * Copyright (c) 2008-2026, Hazelcast, Inc. All Rights Reserved. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -21,7 +21,7 @@ const { Client } = require('hazelcast-client'); const client = await Client.newHazelcastClient(); const topic = await client.getReliableTopic('my-distributed-topic'); - topic.addMessageListener((message) => { + await topic.addMessageListener((message) => { console.log('Received message:\n', message); // shut down the client once the message is received client.shutdown().catch((err) => { diff --git a/code_samples/topic.js b/code_samples/topic.js new file mode 100644 index 000000000..c3c1d5d9a --- /dev/null +++ b/code_samples/topic.js @@ -0,0 +1,36 @@ +/* + * Copyright (c) 2008-2026, Hazelcast, Inc. All Rights Reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +'use strict'; + +const { Client } = require('hazelcast-client'); + +(async () => { + const client = await Client.newHazelcastClient(); + const topic = await client.getTopic('my-distributed-topic'); + + topic.addMessageListener((message) => { + console.log('Received message:\n', message); + // shut down the client once the message is received + client.shutdown().catch((err) => { + console.error('Failed to shut down the client:', err); + }); + }); + + await topic.publish('Hello to distributed world'); +})().catch(err => { + console.error('Error occurred:', err); + process.exit(1); +}); diff --git a/src/HazelcastClient.ts b/src/HazelcastClient.ts index ce0eaef54..8e172fb53 100644 --- a/src/HazelcastClient.ts +++ b/src/HazelcastClient.ts @@ -1,5 +1,5 @@ /* - * Copyright (c) 2008-2022, Hazelcast, Inc. All Rights Reserved. + * Copyright (c) 2008-2026, Hazelcast, Inc. All Rights Reserved. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -402,6 +402,13 @@ export class HazelcastClient { return this.proxyManager.getOrCreateProxy(name, ProxyManager.RELIABLETOPIC_SERVICE) as Promise>; } + /** + * Returns a distributed Topic instance with the given name. + */ + getTopic(name: string): Promise> { + return this.proxyManager.getOrCreateProxy(name, ProxyManager.TOPIC_SERVICE) as Promise>; + } + /** * Returns the distributed Replicated Map instance with given name. */ diff --git a/src/codec/TopicAddMessageListenerCodec.ts b/src/codec/TopicAddMessageListenerCodec.ts new file mode 100644 index 000000000..c87802e28 --- /dev/null +++ b/src/codec/TopicAddMessageListenerCodec.ts @@ -0,0 +1,80 @@ +/* + * Copyright (c) 2008-2022, Hazelcast, Inc. All Rights Reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +/** + * Auto-generated by Hazelcast Client Protocol Code Generator. + * + * @generated Hazelcast Client Protocol Code Generator @e5b5a6b + * @see Protocol Generator + * DO NOT MODIFY. Changes to this file will be lost when the code is regenerated. + */ + +/* eslint-disable max-len */ +import {BitsUtil} from '../util/BitsUtil'; +import {FixSizedTypesCodec} from './builtin/FixSizedTypesCodec'; +import {ClientMessage, Frame, RESPONSE_BACKUP_ACKS_OFFSET, PARTITION_ID_OFFSET} from '../protocol/ClientMessage'; +import {StringCodec} from './builtin/StringCodec'; +import {UUID} from '../core/UUID'; +import * as Long from 'long'; +import {Data} from '../serialization/Data'; +import {DataCodec} from './builtin/DataCodec'; + +// hex: 0x040200 +const REQUEST_MESSAGE_TYPE = 262656; +// hex: 0x040201 +// RESPONSE_MESSAGE_TYPE = 262657 +// hex: 0x040202 +const EVENT_TOPIC_MESSAGE_TYPE = 262658; + +const REQUEST_LOCAL_ONLY_OFFSET = PARTITION_ID_OFFSET + BitsUtil.INT_SIZE_IN_BYTES; +const REQUEST_INITIAL_FRAME_SIZE = REQUEST_LOCAL_ONLY_OFFSET + BitsUtil.BOOLEAN_SIZE_IN_BYTES; +const RESPONSE_RESPONSE_OFFSET = RESPONSE_BACKUP_ACKS_OFFSET + BitsUtil.BYTE_SIZE_IN_BYTES; +const EVENT_TOPIC_PUBLISH_TIME_OFFSET = PARTITION_ID_OFFSET + BitsUtil.INT_SIZE_IN_BYTES; +const EVENT_TOPIC_UUID_OFFSET = EVENT_TOPIC_PUBLISH_TIME_OFFSET + BitsUtil.LONG_SIZE_IN_BYTES; + +/** @internal */ +export class TopicAddMessageListenerCodec { + static encodeRequest(name: string, localOnly: boolean): ClientMessage { + const clientMessage = ClientMessage.createForEncode(); + clientMessage.setRetryable(false); + + const initialFrame = Frame.createInitialFrame(REQUEST_INITIAL_FRAME_SIZE); + FixSizedTypesCodec.encodeBoolean(initialFrame.content, REQUEST_LOCAL_ONLY_OFFSET, localOnly); + clientMessage.addFrame(initialFrame); + clientMessage.setMessageType(REQUEST_MESSAGE_TYPE); + clientMessage.setPartitionId(-1); + + StringCodec.encode(clientMessage, name); + return clientMessage; + } + + static decodeResponse(clientMessage: ClientMessage): UUID { + const initialFrame = clientMessage.nextFrame(); + + return FixSizedTypesCodec.decodeUUID(initialFrame.content, RESPONSE_RESPONSE_OFFSET); + } + + static handle(clientMessage: ClientMessage, handleTopicEvent: (item: Data, publishTime: Long, uuid: UUID) => void = null): void { + const messageType = clientMessage.getMessageType(); + if (messageType === EVENT_TOPIC_MESSAGE_TYPE && handleTopicEvent !== null) { + const initialFrame = clientMessage.nextFrame(); + const publishTime = FixSizedTypesCodec.decodeLong(initialFrame.content, EVENT_TOPIC_PUBLISH_TIME_OFFSET); + const uuid = FixSizedTypesCodec.decodeUUID(initialFrame.content, EVENT_TOPIC_UUID_OFFSET); + const item = DataCodec.decode(clientMessage); + handleTopicEvent(item, publishTime, uuid); + return; + } + } +} diff --git a/src/codec/TopicPublishAllCodec.ts b/src/codec/TopicPublishAllCodec.ts new file mode 100644 index 000000000..ff55703df --- /dev/null +++ b/src/codec/TopicPublishAllCodec.ts @@ -0,0 +1,55 @@ +/* + * Copyright (c) 2008-2022, Hazelcast, Inc. All Rights Reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +/** + * Auto-generated by Hazelcast Client Protocol Code Generator. + * + * @generated Hazelcast Client Protocol Code Generator @e5b5a6b + * @see Protocol Generator + * DO NOT MODIFY. Changes to this file will be lost when the code is regenerated. + */ + +/* eslint-disable max-len */ +import {BitsUtil} from '../util/BitsUtil'; +import {ClientMessage, Frame, PARTITION_ID_OFFSET} from '../protocol/ClientMessage'; +import {StringCodec} from './builtin/StringCodec'; +import {Data} from '../serialization/Data'; +import {ListMultiFrameCodec} from './builtin/ListMultiFrameCodec'; +import {DataCodec} from './builtin/DataCodec'; + +// hex: 0x040400 +const REQUEST_MESSAGE_TYPE = 263168; +// hex: 0x040401 +// RESPONSE_MESSAGE_TYPE = 263169 + +const REQUEST_INITIAL_FRAME_SIZE = PARTITION_ID_OFFSET + BitsUtil.INT_SIZE_IN_BYTES; + +/** @internal */ +export class TopicPublishAllCodec { + static encodeRequest(name: string, messages: Data[]): ClientMessage { + const clientMessage = ClientMessage.createForEncode(); + clientMessage.setRetryable(false); + clientMessage.setContainsSerializedDataInRequest(true); + + const initialFrame = Frame.createInitialFrame(REQUEST_INITIAL_FRAME_SIZE); + clientMessage.addFrame(initialFrame); + clientMessage.setMessageType(REQUEST_MESSAGE_TYPE); + clientMessage.setPartitionId(-1); + + StringCodec.encode(clientMessage, name); + ListMultiFrameCodec.encode(clientMessage, messages, DataCodec.encode); + return clientMessage; + } +} diff --git a/src/codec/TopicPublishCodec.ts b/src/codec/TopicPublishCodec.ts new file mode 100644 index 000000000..857d500cd --- /dev/null +++ b/src/codec/TopicPublishCodec.ts @@ -0,0 +1,54 @@ +/* + * Copyright (c) 2008-2022, Hazelcast, Inc. All Rights Reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +/** + * Auto-generated by Hazelcast Client Protocol Code Generator. + * + * @generated Hazelcast Client Protocol Code Generator @e5b5a6b + * @see Protocol Generator + * DO NOT MODIFY. Changes to this file will be lost when the code is regenerated. + */ + +/* eslint-disable max-len */ +import {BitsUtil} from '../util/BitsUtil'; +import {ClientMessage, Frame, PARTITION_ID_OFFSET} from '../protocol/ClientMessage'; +import {StringCodec} from './builtin/StringCodec'; +import {Data} from '../serialization/Data'; +import {DataCodec} from './builtin/DataCodec'; + +// hex: 0x040100 +const REQUEST_MESSAGE_TYPE = 262400; +// hex: 0x040101 +// RESPONSE_MESSAGE_TYPE = 262401 + +const REQUEST_INITIAL_FRAME_SIZE = PARTITION_ID_OFFSET + BitsUtil.INT_SIZE_IN_BYTES; + +/** @internal */ +export class TopicPublishCodec { + static encodeRequest(name: string, message: Data): ClientMessage { + const clientMessage = ClientMessage.createForEncode(); + clientMessage.setRetryable(false); + clientMessage.setContainsSerializedDataInRequest(true); + + const initialFrame = Frame.createInitialFrame(REQUEST_INITIAL_FRAME_SIZE); + clientMessage.addFrame(initialFrame); + clientMessage.setMessageType(REQUEST_MESSAGE_TYPE); + clientMessage.setPartitionId(-1); + + StringCodec.encode(clientMessage, name); + DataCodec.encode(clientMessage, message); + return clientMessage; + } +} diff --git a/src/codec/TopicRemoveMessageListenerCodec.ts b/src/codec/TopicRemoveMessageListenerCodec.ts new file mode 100644 index 000000000..82fef7caf --- /dev/null +++ b/src/codec/TopicRemoveMessageListenerCodec.ts @@ -0,0 +1,61 @@ +/* + * Copyright (c) 2008-2022, Hazelcast, Inc. All Rights Reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +/** + * Auto-generated by Hazelcast Client Protocol Code Generator. + * + * @generated Hazelcast Client Protocol Code Generator @e5b5a6b + * @see Protocol Generator + * DO NOT MODIFY. Changes to this file will be lost when the code is regenerated. + */ + +/* eslint-disable max-len */ +import {BitsUtil} from '../util/BitsUtil'; +import {FixSizedTypesCodec} from './builtin/FixSizedTypesCodec'; +import {ClientMessage, Frame, RESPONSE_BACKUP_ACKS_OFFSET, PARTITION_ID_OFFSET} from '../protocol/ClientMessage'; +import {UUID} from '../core/UUID'; +import {StringCodec} from './builtin/StringCodec'; + +// hex: 0x040300 +const REQUEST_MESSAGE_TYPE = 262912; +// hex: 0x040301 +// RESPONSE_MESSAGE_TYPE = 262913 + +const REQUEST_REGISTRATION_ID_OFFSET = PARTITION_ID_OFFSET + BitsUtil.INT_SIZE_IN_BYTES; +const REQUEST_INITIAL_FRAME_SIZE = REQUEST_REGISTRATION_ID_OFFSET + BitsUtil.UUID_SIZE_IN_BYTES; +const RESPONSE_RESPONSE_OFFSET = RESPONSE_BACKUP_ACKS_OFFSET + BitsUtil.BYTE_SIZE_IN_BYTES; + +/** @internal */ +export class TopicRemoveMessageListenerCodec { + static encodeRequest(name: string, registrationId: UUID): ClientMessage { + const clientMessage = ClientMessage.createForEncode(); + clientMessage.setRetryable(true); + + const initialFrame = Frame.createInitialFrame(REQUEST_INITIAL_FRAME_SIZE); + FixSizedTypesCodec.encodeUUID(initialFrame.content, REQUEST_REGISTRATION_ID_OFFSET, registrationId); + clientMessage.addFrame(initialFrame); + clientMessage.setMessageType(REQUEST_MESSAGE_TYPE); + clientMessage.setPartitionId(-1); + + StringCodec.encode(clientMessage, name); + return clientMessage; + } + + static decodeResponse(clientMessage: ClientMessage): boolean { + const initialFrame = clientMessage.nextFrame(); + + return FixSizedTypesCodec.decodeBoolean(initialFrame.content, RESPONSE_RESPONSE_OFFSET); + } +} diff --git a/src/proxy/ITopic.ts b/src/proxy/ITopic.ts index fa282f46f..8905d95b1 100644 --- a/src/proxy/ITopic.ts +++ b/src/proxy/ITopic.ts @@ -1,5 +1,5 @@ /* - * Copyright (c) 2008-2022, Hazelcast, Inc. All Rights Reserved. + * Copyright (c) 2008-2026, Hazelcast, Inc. All Rights Reserved. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -22,7 +22,9 @@ import {MessageListener} from './MessageListener'; * delivered to multiple subscribers, which is also known as a publish/subscribe * (pub/sub) messaging model. Publish and subscriptions are cluster-wide. * - * This interface stand for reliable topic, i.e. it uses a Ringbuffer to store + * This interface stand for Topic and Reliable Topic. + * + * ReliableTopic uses a Ringbuffer to store * events. The events in the Ringbuffer are replicated, so they won't get * lost when a node goes down. * @@ -41,7 +43,7 @@ export interface ITopic extends DistributedObject { * @param listener the MessageListener to add * @return registration ID */ - addMessageListener(listener: MessageListener): string; + addMessageListener(listener: MessageListener): Promise; /** * Stops receiving messages for the given message listener. @@ -51,7 +53,7 @@ export interface ITopic extends DistributedObject { * @param listenerId listener registration ID * @return `true` if registration is removed, `false` otherwise */ - removeMessageListener(listenerId: string): boolean; + removeMessageListener(listenerId: string): Promise; /** * Publishes the message to all subscribers of this topic. diff --git a/src/proxy/ProxyManager.ts b/src/proxy/ProxyManager.ts index adfe929c9..fa36b3e6b 100644 --- a/src/proxy/ProxyManager.ts +++ b/src/proxy/ProxyManager.ts @@ -29,6 +29,7 @@ import {ReplicatedMapProxy} from './ReplicatedMapProxy'; import {RingbufferProxy} from './ringbuffer/RingbufferProxy'; import {SetProxy} from './SetProxy'; import {ReliableTopicProxy} from './topic/ReliableTopicProxy'; +import {TopicProxy} from './topic/TopicProxy'; import {deferredPromise} from '../util/Util'; import {ClientMessage} from '../protocol/ClientMessage'; import {ClientCreateProxiesCodec} from '../codec/ClientCreateProxiesCodec'; @@ -68,6 +69,7 @@ export class ProxyManager { public static readonly FLAKEID_SERVICE: string = 'hz:impl:flakeIdGeneratorService'; public static readonly PNCOUNTER_SERVICE: string = 'hz:impl:PNCounterService'; public static readonly RELIABLETOPIC_SERVICE: string = 'hz:impl:reliableTopicService'; + public static readonly TOPIC_SERVICE: string = 'hz:impl:topicService'; public readonly service: { [serviceName: string]: any } = {}; private readonly proxies = new Map>(); @@ -98,6 +100,7 @@ export class ProxyManager { this.service[ProxyManager.FLAKEID_SERVICE] = FlakeIdGeneratorProxy; this.service[ProxyManager.PNCOUNTER_SERVICE] = PNCounterProxy; this.service[ProxyManager.RELIABLETOPIC_SERVICE] = ReliableTopicProxy; + this.service[ProxyManager.TOPIC_SERVICE] = TopicProxy; } public getOrCreateProxy(name: string, serviceName: string, createAtServer = true): Promise { diff --git a/src/proxy/topic/ReliableTopicListenerRunner.ts b/src/proxy/topic/ReliableTopicListenerRunner.ts index 331925930..3e12eadc9 100644 --- a/src/proxy/topic/ReliableTopicListenerRunner.ts +++ b/src/proxy/topic/ReliableTopicListenerRunner.ts @@ -92,7 +92,11 @@ export class ReliableTopicListenerRunner { if (this.handleInternalError(err)) { setImmediate(this.next.bind(this)); } else { - this.proxy.removeMessageListener(this.listenerId); + // ReliableTopic removeMessageListener never returns an error with removeMessageListener + // handling the error for completeness --YT + this.proxy.removeMessageListener(this.listenerId).catch((err) => { + throw err; + }); } }); } diff --git a/src/proxy/topic/ReliableTopicProxy.ts b/src/proxy/topic/ReliableTopicProxy.ts index 01b003028..248836ee2 100644 --- a/src/proxy/topic/ReliableTopicProxy.ts +++ b/src/proxy/topic/ReliableTopicProxy.ts @@ -92,7 +92,7 @@ export class ReliableTopicProxy extends BaseProxy implements ITopic { this.ringbuffer = ringbuffer; } - addMessageListener(listener: MessageListener): string { + addMessageListener(listener: MessageListener): Promise { const listenerId = UuidUtil.generate().toString(); const runner = new ReliableTopicListenerRunner(listenerId, listener, this.ringbuffer, this.batchSize, this.serializationService, this.logger, this); @@ -106,19 +106,19 @@ export class ReliableTopicProxy extends BaseProxy implements ITopic { this.logger.warn('ReliableTopicProxy', 'Failed to fetch sequence for runner.', e); }); - return listenerId; + return Promise.resolve(listenerId); } - removeMessageListener(listenerId: string): boolean { + removeMessageListener(listenerId: string): Promise { const runner = this.runners[listenerId]; if (!runner) { - return false; + return Promise.resolve(false); } runner.cancel(); delete this.runners[listenerId]; - return true; + return Promise.resolve(true); } publish(message: E): Promise { diff --git a/src/proxy/topic/TopicProxy.ts b/src/proxy/topic/TopicProxy.ts new file mode 100644 index 000000000..c1bf26f9d --- /dev/null +++ b/src/proxy/topic/TopicProxy.ts @@ -0,0 +1,81 @@ +/* + * Copyright (c) 2008-2026, Hazelcast, Inc. All Rights Reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +/** @ignore *//** */ + +import {PartitionSpecificProxy} from '../PartitionSpecificProxy'; +import {ITopic} from '../ITopic'; +import {SchemaNotReplicatedError, UUID} from '../../core'; +import {MessageListener, Message} from '../MessageListener'; +import {TopicPublishCodec} from '../../codec/TopicPublishCodec'; +import {Data} from '../../serialization'; +import {ClientMessage} from '../../protocol/ClientMessage'; +import {TopicAddMessageListenerCodec} from '../../codec/TopicAddMessageListenerCodec'; +import {ListenerMessageCodec} from '../../listener/ListenerMessageCodec'; +import {TopicRemoveMessageListenerCodec} from '../../codec/TopicRemoveMessageListenerCodec'; + +export class TopicProxy extends PartitionSpecificProxy implements ITopic { + addMessageListener(listener: MessageListener): Promise { + const handler = (message: ClientMessage): void => { + TopicAddMessageListenerCodec.handle(message, (item, publishTime, uuid) => { + let responseObject: E; + if (item == null) { + responseObject = null; + } else { + responseObject = this.toObject(item); + } + const member = this.clusterService.getMember(uuid.toString()); + const message = new Message(); + message.messageObject = responseObject; + message.publisher = member.address; + message.publishingTime = publishTime; + listener.apply(null, [message]); + }); + } + const codec = this.createEntryListener(this.name); + return this.listenerService.registerListener(codec, handler); + } + + publish(message: E): Promise { + let messageData: Data; + try { + messageData = this.toData(message); + } catch (e) { + if (e instanceof SchemaNotReplicatedError) { + return this.registerSchema(e.schema, e.clazz).then(() => this.publish(message)); + } + throw e; + } + return this.encodeInvoke(TopicPublishCodec, () => {}, messageData); + } + + removeMessageListener(listenerId: string): Promise { + return this.listenerService.deregisterListener(listenerId); + } + + private createEntryListener(name: string): ListenerMessageCodec { + return { + encodeAddRequest(localOnly: boolean): ClientMessage { + return TopicAddMessageListenerCodec.encodeRequest(name, localOnly); + }, + decodeAddResponse(msg: ClientMessage): UUID { + return TopicAddMessageListenerCodec.decodeResponse(msg); + }, + encodeRemoveRequest(listenerId: UUID): ClientMessage { + return TopicRemoveMessageListenerCodec.encodeRequest(name, listenerId); + }, + }; + } +} diff --git a/test/integration/backward_compatible/parallel/serialization/compact/CompactPublicAPIsTest.js b/test/integration/backward_compatible/parallel/serialization/compact/CompactPublicAPIsTest.js index cf113b3ca..43bb3eb1f 100644 --- a/test/integration/backward_compatible/parallel/serialization/compact/CompactPublicAPIsTest.js +++ b/test/integration/backward_compatible/parallel/serialization/compact/CompactPublicAPIsTest.js @@ -161,7 +161,7 @@ describe('CompactPublicAPIsTest', function () { let member; let client; let map, multimap, list, atomicReference, - queue, set, topic, nearCachedMap1, nearCachedMap2, + queue, set, reliableTopic, topic, nearCachedMap1, nearCachedMap2, replicatedMap, ringBuffer; let employee; let SchemaNotReplicatedError; @@ -172,7 +172,7 @@ describe('CompactPublicAPIsTest', function () { + http://www.hazelcast.com/schema/config/hazelcast-config-5.7.xsd"> 0 @@ -260,7 +260,8 @@ describe('CompactPublicAPIsTest', function () { } queue = await client.getQueue(name); set = await client.getSet(name); - topic = await client.getReliableTopic(name); + reliableTopic = await client.getReliableTopic(name); + topic = await client.getTopic(name); ringBuffer = await client.getRingbuffer(name); compactSerializerUsed = false; }); @@ -281,6 +282,7 @@ describe('CompactPublicAPIsTest', function () { } await queue.destroy(); await set.destroy(); + await reliableTopic.destroy(); await topic.destroy(); await ringBuffer.destroy(); await testFactory.shutdownAllClients(); @@ -1067,6 +1069,14 @@ describe('CompactPublicAPIsTest', function () { }); describe('ReliableTopic', function () { + it('publish', async function () { + const fn = reliableTopic.publish.bind(reliableTopic, OUTER_INSTANCE); + await fn(); + shouldThrowSerializationErrors(client, fn); + }); + }); + + describe('Topic', function () { it('publish', async function () { const fn = topic.publish.bind(topic, OUTER_INSTANCE); await fn(); diff --git a/test/integration/backward_compatible/parallel/topic/ReliableTopicTest.js b/test/integration/backward_compatible/parallel/topic/ReliableTopicTest.js index 58847a0d9..27ba98cc3 100644 --- a/test/integration/backward_compatible/parallel/topic/ReliableTopicTest.js +++ b/test/integration/backward_compatible/parallel/topic/ReliableTopicTest.js @@ -103,22 +103,25 @@ describe('ReliableTopicTest', function () { }).then((topic) => { topicTwo = topic; let receivedMessages = 0; - const id = topicTwo.addMessageListener(() => { + let id = null; + topicTwo.addMessageListener(() => { receivedMessages++; if (receivedMessages > 2) { done(new Error('Kept receiving messages after message listener is removed.')); } + }).then((listenerId) => { + id = listenerId; }); - topicOne.publish({ 'value0': 'foo0' }); topicOne.publish({ 'value1': 'foo1' }); setTimeout(() => { - topicTwo.removeMessageListener(id); - topicOne.publish({ 'value2': 'foo2' }); - topicOne.publish({ 'value3': 'foo3' }); - topicOne.publish({ 'value4': 'foo4' }); - topicOne.publish({ 'value5': 'foo5' }); - setTimeout(done, 500); + topicTwo.removeMessageListener(id).then(() => { + topicOne.publish({ 'value2': 'foo2' }); + topicOne.publish({ 'value3': 'foo3' }); + topicOne.publish({ 'value4': 'foo4' }); + topicOne.publish({ 'value5': 'foo5' }); + setTimeout(done, 500); + }); }, 500); }).catch(done); }); diff --git a/test/integration/backward_compatible/parallel/topic/TopicTest.js b/test/integration/backward_compatible/parallel/topic/TopicTest.js new file mode 100644 index 000000000..cdca0ed0b --- /dev/null +++ b/test/integration/backward_compatible/parallel/topic/TopicTest.js @@ -0,0 +1,84 @@ +/* + * Copyright (c) 2008-2026, Hazelcast, Inc. All Rights Reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +'use strict'; + +const chai = require('chai'); +const expect = chai.expect; +const chaiAsPromised = require('chai-as-promised'); +chai.use(chaiAsPromised); + +const RC = require('../../../RC'); +const TestUtil = require('../../../../TestUtil'); + +describe('TopicProxyTest', function () { + let cluster; + let client; + let topic; + const testFactory = new TestUtil.TestFactory(); + + before(async function () { + cluster = await testFactory.createClusterForParallelTests(); + const member = await RC.startMember(cluster.id); + client = await testFactory.newHazelcastClientForParallelTests({ clusterName: cluster.id }, member); + }); + + beforeEach(async function () { + topic = await client.getTopic('ClientTopicTest'); + return null; + }); + + afterEach(async function () { + return topic.destroy(); + }); + + after(async function () { + await testFactory.shutdownAll(); + }); + + it('should be able to add a listener', async function() { + const target = 'item-value'; + const collector = eventCollector(); + const topic = await client.getTopic('topic-listener'); + await topic.addMessageListener(collector); + await topic.publish(target); + await TestUtil.assertTrueEventually(async () => { + expect(collector.events.length).equal(1); + const event = collector.events[0]; + expect(event.messageObject).equal(target); + expect(event.publishingTime.isZero()).equal(false); + }); + }); + + it('should be able to remove a listener', async function() { + const collector = eventCollector(); + const topic = await client.getTopic('topic-listener'); + const listenerId = await topic.addMessageListener(collector); + await topic.removeMessageListener(listenerId); + await topic.publish('item-value'); + await TestUtil.assertTrueEventually(async () => { + expect(collector.events.length).equal(0); + }); + }); +}); + +function eventCollector() { + const events = []; + function collector(e) { + events.push(e); + } + collector.events = events; + return collector; +}