[HZ-5498] Topic Implementation

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.

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) => {

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);
+});

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<ITopic<E>>;
     }
 
+    /**
+     * Returns a distributed Topic instance with the given name.
+     */
+    getTopic<E>(name: string): Promise<ITopic<E>> {
+        return this.proxyManager.getOrCreateProxy(name, ProxyManager.TOPIC_SERVICE) as Promise<ITopic<E>>;
+    }
+
     /**
      * Returns the distributed Replicated Map instance with given name.
      */

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 <a href="https://github.com/hazelcast/hazelcast-client-protocol">Protocol Generator</a>
+ * 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;
+        }
+    }
+}

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 <a href="https://github.com/hazelcast/hazelcast-client-protocol">Protocol Generator</a>
+ * 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;
+    }
+}

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 <a href="https://github.com/hazelcast/hazelcast-client-protocol">Protocol Generator</a>
+ * 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;
+    }
+}

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 <a href="https://github.com/hazelcast/hazelcast-client-protocol">Protocol Generator</a>
+ * 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);
+    }
+}