Skip to content

Commit a756754

Browse files
yuceyuce
authored
[HZ-5498] Topic Implementation (#1604)
* Added `Topic` data structure. * Updated `Topic` interface to fit both `Topic` and `ReliableTopic`.
1 parent c487281 commit a756754

16 files changed

Lines changed: 528 additions & 25 deletions

‎DOCUMENTATION.md‎

Lines changed: 24 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2084,6 +2084,29 @@ of these policies.
20842084
> **NOTE: When you use `default` as the Reliable Topic configuration key, it has a special meaning. Hazelcast client will use
20852085
> that configuration as the default one for all Reliable Topics, unless there is a specific configuration for the topic.**
20862086
2087+
2088+
### 8.4.8. Using Topic
2089+
2090+
Hazelcast `Topic` is a distributed topic implementation. For details, see
2091+
the [Topic section](https://docs.hazelcast.com/hazelcast/latest/data-structures/topic) in the Hazelcast
2092+
Reference Manual.
2093+
2094+
A Topic usage example is shown below.
2095+
2096+
```javascript
2097+
// Get a Topic called 'my-distributed-topic'
2098+
const topic = await client.getTopic('my-distributed-topic');
2099+
// Add a Listener to the Topic
2100+
topic.addMessageListener((message) => {
2101+
console.log('Message:', message);
2102+
});
2103+
// Publish a message to the Topic
2104+
await topic.publish('Hello to the distributed world!');
2105+
```
2106+
2107+
Hazelcast Topic uses `MessageListener` to listen to the events that occur when a message is received. See the
2108+
[Message Listener section](#8524-message-listener) for information on how to create a message listener object and register it.
2109+
20872110
### 8.4.9. Using PN Counter
20882111

20892112
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.
27312754

27322755
#### 8.5.2.4. Message Listener
27332756

2734-
The Message Listener is used by the Hazelcast `ReliableTopic`.
2757+
The Message Listener is used by the Hazelcast `Topic` and `ReliableTopic`.
27352758

27362759
You can listen to message events. To listen to these events, you need to implement the `MessageListener` function to which a
27372760
`Message` object is passed.

‎code_samples/reliable_topic.js‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright (c) 2008-2022, Hazelcast, Inc. All Rights Reserved.
2+
* Copyright (c) 2008-2026, Hazelcast, Inc. All Rights Reserved.
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -21,7 +21,7 @@ const { Client } = require('hazelcast-client');
2121
const client = await Client.newHazelcastClient();
2222
const topic = await client.getReliableTopic('my-distributed-topic');
2323

24-
topic.addMessageListener((message) => {
24+
await topic.addMessageListener((message) => {
2525
console.log('Received message:\n', message);
2626
// shut down the client once the message is received
2727
client.shutdown().catch((err) => {

‎code_samples/topic.js‎

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,36 @@
1+
/*
2+
* Copyright (c) 2008-2026, Hazelcast, Inc. All Rights Reserved.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
'use strict';
17+
18+
const { Client } = require('hazelcast-client');
19+
20+
(async () => {
21+
const client = await Client.newHazelcastClient();
22+
const topic = await client.getTopic('my-distributed-topic');
23+
24+
topic.addMessageListener((message) => {
25+
console.log('Received message:\n', message);
26+
// shut down the client once the message is received
27+
client.shutdown().catch((err) => {
28+
console.error('Failed to shut down the client:', err);
29+
});
30+
});
31+
32+
await topic.publish('Hello to distributed world');
33+
})().catch(err => {
34+
console.error('Error occurred:', err);
35+
process.exit(1);
36+
});

‎src/HazelcastClient.ts‎

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright (c) 2008-2022, Hazelcast, Inc. All Rights Reserved.
2+
* Copyright (c) 2008-2026, Hazelcast, Inc. All Rights Reserved.
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -402,6 +402,13 @@ export class HazelcastClient {
402402
return this.proxyManager.getOrCreateProxy(name, ProxyManager.RELIABLETOPIC_SERVICE) as Promise<ITopic<E>>;
403403
}
404404

405+
/**
406+
* Returns a distributed Topic instance with the given name.
407+
*/
408+
getTopic<E>(name: string): Promise<ITopic<E>> {
409+
return this.proxyManager.getOrCreateProxy(name, ProxyManager.TOPIC_SERVICE) as Promise<ITopic<E>>;
410+
}
411+
405412
/**
406413
* Returns the distributed Replicated Map instance with given name.
407414
*/

‎src/codec/TopicAddMessageListenerCodec.ts‎

Lines changed: 80 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,80 @@
1+
/*
2+
* Copyright (c) 2008-2022, Hazelcast, Inc. All Rights Reserved.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
/**
17+
* Auto-generated by Hazelcast Client Protocol Code Generator.
18+
*
19+
* @generated Hazelcast Client Protocol Code Generator @e5b5a6b
20+
* @see <a href="https://github.com/hazelcast/hazelcast-client-protocol">Protocol Generator</a>
21+
* DO NOT MODIFY. Changes to this file will be lost when the code is regenerated.
22+
*/
23+
24+
/* eslint-disable max-len */
25+
import {BitsUtil} from '../util/BitsUtil';
26+
import {FixSizedTypesCodec} from './builtin/FixSizedTypesCodec';
27+
import {ClientMessage, Frame, RESPONSE_BACKUP_ACKS_OFFSET, PARTITION_ID_OFFSET} from '../protocol/ClientMessage';
28+
import {StringCodec} from './builtin/StringCodec';
29+
import {UUID} from '../core/UUID';
30+
import * as Long from 'long';
31+
import {Data} from '../serialization/Data';
32+
import {DataCodec} from './builtin/DataCodec';
33+
34+
// hex: 0x040200
35+
const REQUEST_MESSAGE_TYPE = 262656;
36+
// hex: 0x040201
37+
// RESPONSE_MESSAGE_TYPE = 262657
38+
// hex: 0x040202
39+
const EVENT_TOPIC_MESSAGE_TYPE = 262658;
40+
41+
const REQUEST_LOCAL_ONLY_OFFSET = PARTITION_ID_OFFSET + BitsUtil.INT_SIZE_IN_BYTES;
42+
const REQUEST_INITIAL_FRAME_SIZE = REQUEST_LOCAL_ONLY_OFFSET + BitsUtil.BOOLEAN_SIZE_IN_BYTES;
43+
const RESPONSE_RESPONSE_OFFSET = RESPONSE_BACKUP_ACKS_OFFSET + BitsUtil.BYTE_SIZE_IN_BYTES;
44+
const EVENT_TOPIC_PUBLISH_TIME_OFFSET = PARTITION_ID_OFFSET + BitsUtil.INT_SIZE_IN_BYTES;
45+
const EVENT_TOPIC_UUID_OFFSET = EVENT_TOPIC_PUBLISH_TIME_OFFSET + BitsUtil.LONG_SIZE_IN_BYTES;
46+
47+
/** @internal */
48+
export class TopicAddMessageListenerCodec {
49+
static encodeRequest(name: string, localOnly: boolean): ClientMessage {
50+
const clientMessage = ClientMessage.createForEncode();
51+
clientMessage.setRetryable(false);
52+
53+
const initialFrame = Frame.createInitialFrame(REQUEST_INITIAL_FRAME_SIZE);
54+
FixSizedTypesCodec.encodeBoolean(initialFrame.content, REQUEST_LOCAL_ONLY_OFFSET, localOnly);
55+
clientMessage.addFrame(initialFrame);
56+
clientMessage.setMessageType(REQUEST_MESSAGE_TYPE);
57+
clientMessage.setPartitionId(-1);
58+
59+
StringCodec.encode(clientMessage, name);
60+
return clientMessage;
61+
}
62+
63+
static decodeResponse(clientMessage: ClientMessage): UUID {
64+
const initialFrame = clientMessage.nextFrame();
65+
66+
return FixSizedTypesCodec.decodeUUID(initialFrame.content, RESPONSE_RESPONSE_OFFSET);
67+
}
68+
69+
static handle(clientMessage: ClientMessage, handleTopicEvent: (item: Data, publishTime: Long, uuid: UUID) => void = null): void {
70+
const messageType = clientMessage.getMessageType();
71+
if (messageType === EVENT_TOPIC_MESSAGE_TYPE && handleTopicEvent !== null) {
72+
const initialFrame = clientMessage.nextFrame();
73+
const publishTime = FixSizedTypesCodec.decodeLong(initialFrame.content, EVENT_TOPIC_PUBLISH_TIME_OFFSET);
74+
const uuid = FixSizedTypesCodec.decodeUUID(initialFrame.content, EVENT_TOPIC_UUID_OFFSET);
75+
const item = DataCodec.decode(clientMessage);
76+
handleTopicEvent(item, publishTime, uuid);
77+
return;
78+
}
79+
}
80+
}

‎src/codec/TopicPublishAllCodec.ts‎

Lines changed: 55 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,55 @@
1+
/*
2+
* Copyright (c) 2008-2022, Hazelcast, Inc. All Rights Reserved.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
/**
17+
* Auto-generated by Hazelcast Client Protocol Code Generator.
18+
*
19+
* @generated Hazelcast Client Protocol Code Generator @e5b5a6b
20+
* @see <a href="https://github.com/hazelcast/hazelcast-client-protocol">Protocol Generator</a>
21+
* DO NOT MODIFY. Changes to this file will be lost when the code is regenerated.
22+
*/
23+
24+
/* eslint-disable max-len */
25+
import {BitsUtil} from '../util/BitsUtil';
26+
import {ClientMessage, Frame, PARTITION_ID_OFFSET} from '../protocol/ClientMessage';
27+
import {StringCodec} from './builtin/StringCodec';
28+
import {Data} from '../serialization/Data';
29+
import {ListMultiFrameCodec} from './builtin/ListMultiFrameCodec';
30+
import {DataCodec} from './builtin/DataCodec';
31+
32+
// hex: 0x040400
33+
const REQUEST_MESSAGE_TYPE = 263168;
34+
// hex: 0x040401
35+
// RESPONSE_MESSAGE_TYPE = 263169
36+
37+
const REQUEST_INITIAL_FRAME_SIZE = PARTITION_ID_OFFSET + BitsUtil.INT_SIZE_IN_BYTES;
38+
39+
/** @internal */
40+
export class TopicPublishAllCodec {
41+
static encodeRequest(name: string, messages: Data[]): ClientMessage {
42+
const clientMessage = ClientMessage.createForEncode();
43+
clientMessage.setRetryable(false);
44+
clientMessage.setContainsSerializedDataInRequest(true);
45+
46+
const initialFrame = Frame.createInitialFrame(REQUEST_INITIAL_FRAME_SIZE);
47+
clientMessage.addFrame(initialFrame);
48+
clientMessage.setMessageType(REQUEST_MESSAGE_TYPE);
49+
clientMessage.setPartitionId(-1);
50+
51+
StringCodec.encode(clientMessage, name);
52+
ListMultiFrameCodec.encode(clientMessage, messages, DataCodec.encode);
53+
return clientMessage;
54+
}
55+
}

‎src/codec/TopicPublishCodec.ts‎

Lines changed: 54 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,54 @@
1+
/*
2+
* Copyright (c) 2008-2022, Hazelcast, Inc. All Rights Reserved.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
/**
17+
* Auto-generated by Hazelcast Client Protocol Code Generator.
18+
*
19+
* @generated Hazelcast Client Protocol Code Generator @e5b5a6b
20+
* @see <a href="https://github.com/hazelcast/hazelcast-client-protocol">Protocol Generator</a>
21+
* DO NOT MODIFY. Changes to this file will be lost when the code is regenerated.
22+
*/
23+
24+
/* eslint-disable max-len */
25+
import {BitsUtil} from '../util/BitsUtil';
26+
import {ClientMessage, Frame, PARTITION_ID_OFFSET} from '../protocol/ClientMessage';
27+
import {StringCodec} from './builtin/StringCodec';
28+
import {Data} from '../serialization/Data';
29+
import {DataCodec} from './builtin/DataCodec';
30+
31+
// hex: 0x040100
32+
const REQUEST_MESSAGE_TYPE = 262400;
33+
// hex: 0x040101
34+
// RESPONSE_MESSAGE_TYPE = 262401
35+
36+
const REQUEST_INITIAL_FRAME_SIZE = PARTITION_ID_OFFSET + BitsUtil.INT_SIZE_IN_BYTES;
37+
38+
/** @internal */
39+
export class TopicPublishCodec {
40+
static encodeRequest(name: string, message: Data): ClientMessage {
41+
const clientMessage = ClientMessage.createForEncode();
42+
clientMessage.setRetryable(false);
43+
clientMessage.setContainsSerializedDataInRequest(true);
44+
45+
const initialFrame = Frame.createInitialFrame(REQUEST_INITIAL_FRAME_SIZE);
46+
clientMessage.addFrame(initialFrame);
47+
clientMessage.setMessageType(REQUEST_MESSAGE_TYPE);
48+
clientMessage.setPartitionId(-1);
49+
50+
StringCodec.encode(clientMessage, name);
51+
DataCodec.encode(clientMessage, message);
52+
return clientMessage;
53+
}
54+
}

‎src/codec/TopicRemoveMessageListenerCodec.ts‎

Lines changed: 61 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,61 @@
1+
/*
2+
* Copyright (c) 2008-2022, Hazelcast, Inc. All Rights Reserved.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
/**
17+
* Auto-generated by Hazelcast Client Protocol Code Generator.
18+
*
19+
* @generated Hazelcast Client Protocol Code Generator @e5b5a6b
20+
* @see <a href="https://github.com/hazelcast/hazelcast-client-protocol">Protocol Generator</a>
21+
* DO NOT MODIFY. Changes to this file will be lost when the code is regenerated.
22+
*/
23+
24+
/* eslint-disable max-len */
25+
import {BitsUtil} from '../util/BitsUtil';
26+
import {FixSizedTypesCodec} from './builtin/FixSizedTypesCodec';
27+
import {ClientMessage, Frame, RESPONSE_BACKUP_ACKS_OFFSET, PARTITION_ID_OFFSET} from '../protocol/ClientMessage';
28+
import {UUID} from '../core/UUID';
29+
import {StringCodec} from './builtin/StringCodec';
30+
31+
// hex: 0x040300
32+
const REQUEST_MESSAGE_TYPE = 262912;
33+
// hex: 0x040301
34+
// RESPONSE_MESSAGE_TYPE = 262913
35+
36+
const REQUEST_REGISTRATION_ID_OFFSET = PARTITION_ID_OFFSET + BitsUtil.INT_SIZE_IN_BYTES;
37+
const REQUEST_INITIAL_FRAME_SIZE = REQUEST_REGISTRATION_ID_OFFSET + BitsUtil.UUID_SIZE_IN_BYTES;
38+
const RESPONSE_RESPONSE_OFFSET = RESPONSE_BACKUP_ACKS_OFFSET + BitsUtil.BYTE_SIZE_IN_BYTES;
39+
40+
/** @internal */
41+
export class TopicRemoveMessageListenerCodec {
42+
static encodeRequest(name: string, registrationId: UUID): ClientMessage {
43+
const clientMessage = ClientMessage.createForEncode();
44+
clientMessage.setRetryable(true);
45+
46+
const initialFrame = Frame.createInitialFrame(REQUEST_INITIAL_FRAME_SIZE);
47+
FixSizedTypesCodec.encodeUUID(initialFrame.content, REQUEST_REGISTRATION_ID_OFFSET, registrationId);
48+
clientMessage.addFrame(initialFrame);
49+
clientMessage.setMessageType(REQUEST_MESSAGE_TYPE);
50+
clientMessage.setPartitionId(-1);
51+
52+
StringCodec.encode(clientMessage, name);
53+
return clientMessage;
54+
}
55+
56+
static decodeResponse(clientMessage: ClientMessage): boolean {
57+
const initialFrame = clientMessage.nextFrame();
58+
59+
return FixSizedTypesCodec.decodeBoolean(initialFrame.content, RESPONSE_RESPONSE_OFFSET);
60+
}
61+
}

0 commit comments

Comments
 (0)