Skip to content

Commit bc4b99c

Browse files
zhouxinyuzhouxinyu
committed
Remove topic concepts from oms
1 parent f4d80ea commit bc4b99c

File tree

14 files changed

+82
-250
lines changed

14 files changed

+82
-250
lines changed

‎openmessaging-api-samples/src/main/java/io/openmessaging/samples/consumer/PushConsumerApp.java

Lines changed: 14 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -38,7 +38,7 @@ public static void main(String[] args) throws OMSResourceNotExistException {
3838
String simpleQueue = "HELLO_QUEUE";
3939
resourceManager.createQueue("NS1", simpleQueue, OMS.newKeyValue());
4040

41-
//This queue doesn't has a source topic, so only the message delivered to the queue directly can
41+
//This queue doesn't has a source queue, so only the message delivered to the queue directly can
4242
//be consumed by this consumer.
4343
consumer.attachQueue(simpleQueue, new MessageListener() {
4444
@Override
@@ -56,14 +56,23 @@ public void onReceived(Message message, Context context) {
5656
//Consume messages from a complex queue.
5757
final PushConsumer anotherConsumer = messagingAccessPoint.createPushConsumer();
5858
{
59-
String complexQueue = "QUEUE_HAS_SOURCE_TOPIC";
60-
String sourceTopic = "SOURCE_TOPIC";
59+
String complexQueue = "QUEUE_WITH_SOURCE_QUEUE";
60+
String sourceQueue = "SOURCE_QUEUE";
6161

6262
//Create the complex queue.
6363
resourceManager.createQueue("NS_01", complexQueue, OMS.newKeyValue());
64-
//Create the source topic.
65-
resourceManager.createTopic("NS_01", sourceTopic, OMS.newKeyValue());
64+
//Create the source queue.
65+
resourceManager.createQueue("NS_01", sourceQueue, OMS.newKeyValue());
6666

67+
anotherConsumer.attachQueue(complexQueue, new MessageListener() {
68+
@Override
69+
public void onReceived(Message message, Context context) {
70+
//The message sent to the sourceQueue will be delivered to anotherConsumer
71+
System.out.println("Received one message: " + message);
72+
context.ack();
73+
}
74+
75+
});
6776
}
6877

6978
Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {

‎openmessaging-api-samples/src/main/java/io/openmessaging/samples/consumer/StreamingConsumerApp.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -39,7 +39,7 @@ public static void main(String[] args) {
3939
}
4040

4141
//Persist the consume offset.
42-
messageIterator.commit(true);
42+
messageIterator.commit();
4343

4444
Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
4545
@Override

‎openmessaging-api-samples/src/main/java/io/openmessaging/samples/producer/AnotherProducerApp.java

Lines changed: 18 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -19,8 +19,8 @@
1919

2020
import io.openmessaging.MessagingAccessPoint;
2121
import io.openmessaging.OMS;
22+
import io.openmessaging.producer.BatchMessageSender;
2223
import io.openmessaging.producer.Producer;
23-
import io.openmessaging.producer.SendResult;
2424
import java.nio.charset.Charset;
2525

2626
public class AnotherProducerApp {
@@ -35,32 +35,32 @@ public static void main(String[] args) {
3535
producer.startup();
3636
System.out.println("Producer startup OK");
3737

38-
Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
39-
@Override
40-
public void run() {
41-
producer.shutdown();
42-
messagingAccessPoint.shutdown();
43-
}
44-
}));
38+
BatchMessageSender batchMessageSender = producer.createBatchMessageSender();
4539

46-
SendResult result = producer.send(producer.createTopicBytesMessage(
47-
"HELLO_TOPIC1", "HELLO_BODY1".getBytes(Charset.forName("UTF-8"))));
40+
batchMessageSender.send(producer.createQueueBytesMessage(
41+
"HELLO_QUEUE1", "HELLO_BODY1".getBytes(Charset.forName("UTF-8"))));
4842

49-
System.out.println("Send first message to topic OK, message id is: " + result.messageId());
50-
51-
producer.send(producer.createTopicBytesMessage(
52-
"HELLO_TOPIC2", "HELLO_BODY2".getBytes(Charset.forName("UTF-8")))
43+
batchMessageSender.send(producer.createQueueBytesMessage(
44+
"HELLO_QUEUE2", "HELLO_BODY2".getBytes(Charset.forName("UTF-8")))
5345
.putUserHeaders("KEY1", 100)
5446
.putUserHeaders("KEY2", 200L)
5547
.putUserHeaders("KEY3", 3.14)
5648
.putUserHeaders("KEY4", "value4")
5749
);
5850

59-
System.out.println("Send second message to topic OK");
60-
61-
producer.send(producer.createQueueBytesMessage(
51+
batchMessageSender.send(producer.createQueueBytesMessage(
6252
"HELLO_QUEUE", "HELLO_BODY".getBytes(Charset.forName("UTF-8"))));
6353

64-
System.out.println("send third message to queue OK");
54+
batchMessageSender.commit();
55+
56+
System.out.println("Send a batch of messages OK");
57+
58+
Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
59+
@Override
60+
public void run() {
61+
producer.shutdown();
62+
messagingAccessPoint.shutdown();
63+
}
64+
}));
6565
}
6666
}

‎openmessaging-api-samples/src/main/java/io/openmessaging/samples/producer/ProducerApp.java

Lines changed: 16 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -37,36 +37,27 @@ public static void main(String[] args) {
3737
producer.startup();
3838
System.out.println("Producer startup OK");
3939

40-
//Add a shutdown hook
41-
Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
42-
@Override
43-
public void run() {
44-
producer.shutdown();
45-
messagingAccessPoint.shutdown();
46-
}
47-
}));
48-
4940
//Sync
5041
{
51-
SendResult sendResult = producer.send(producer.createTopicBytesMessage(
52-
"HELLO_TOPIC", "HELLO_BODY".getBytes(Charset.forName("UTF-8"))));
42+
SendResult sendResult = producer.send(producer.createQueueBytesMessage(
43+
"HELLO_QUEUE", "HELLO_BODY".getBytes(Charset.forName("UTF-8"))));
5344

5445
System.out.println("Send sync message OK, message id is: " + sendResult.messageId());
5546
}
5647

5748
//Async with Promise
5849
{
59-
final Future<SendResult> result = producer.sendAsync(producer.createTopicBytesMessage(
60-
"HELLO_TOPIC", "HELLO_BODY".getBytes(Charset.forName("UTF-8"))));
50+
final Future<SendResult> result = producer.sendAsync(producer.createQueueBytesMessage(
51+
"HELLO_QUEUE", "HELLO_BODY".getBytes(Charset.forName("UTF-8"))));
6152

6253
final SendResult sendResult = result.get(3000L);
6354
System.out.println("Send async message OK, message id is: " + sendResult.messageId());
6455
}
6556

6657
//Async with FutureListener
6758
{
68-
final Future<SendResult> result = producer.sendAsync(producer.createTopicBytesMessage(
69-
"HELLO_TOPIC", "HELLO_BODY".getBytes(Charset.forName("UTF-8"))));
59+
final Future<SendResult> result = producer.sendAsync(producer.createQueueBytesMessage(
60+
"HELLO_QUEUE", "HELLO_BODY".getBytes(Charset.forName("UTF-8"))));
7061

7162
result.addListener(new FutureListener<SendResult>() {
7263
@Override
@@ -83,10 +74,18 @@ public void operationFailed(Future<SendResult> promise) {
8374

8475
//Oneway
8576
{
86-
producer.sendOneway(producer.createTopicBytesMessage(
87-
"HELLO_TOPIC", "HELLO_BODY".getBytes(Charset.forName("UTF-8"))));
77+
producer.sendOneway(producer.createQueueBytesMessage(
78+
"HELLO_QUEUE", "HELLO_BODY".getBytes(Charset.forName("UTF-8"))));
8879

8980
System.out.println("Send oneway message OK");
9081
}
82+
83+
Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
84+
@Override
85+
public void run() {
86+
producer.shutdown();
87+
messagingAccessPoint.shutdown();
88+
}
89+
}));
9190
}
9291
}

‎openmessaging-api/README.md

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,9 @@
11
## Server Side
22
* Namespace
3-
* Topic
43
* Queue
54
* Stream
65
* Routing
7-
* Operator
6+
* Expression
87

98
## Client Side
109
* Producer

‎openmessaging-api/src/main/java/io/openmessaging/Message.java

Lines changed: 1 addition & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -131,16 +131,6 @@ interface BuiltinKeys {
131131
*/
132132
String MessageId = "MessageId";
133133

134-
/**
135-
* The {@code Topic} header field is the destination which the message is being sent.
136-
* <p>
137-
* When a message is sent this value is should be set properly.
138-
* <p>
139-
* When a message is received, its {@code Topic} value must be equivalent to the
140-
* value assigned when it was sent.
141-
*/
142-
String Topic = "Topic";
143-
144134
/**
145135
* The {@code Queue} header field is the destination which the message is being sent.
146136
* <p>
@@ -284,6 +274,6 @@ interface BuiltinKeys {
284274
*/
285275
String RetryReason = "RetryReason";
286276

287-
String Stream = "Stream";
277+
String StreamName = "StreamName";
288278
}
289279
}

‎openmessaging-api/src/main/java/io/openmessaging/MessageFactory.java

Lines changed: 0 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -24,19 +24,6 @@
2424
*
2525
*/
2626
public interface MessageFactory {
27-
/**
28-
* Creates a {@code BytesMessage} object. A {@code BytesMessage} object is used to send a message containing a
29-
* stream of uninterpreted bytes.
30-
* <p>
31-
* The returned {@code BytesMessage} object only can be sent to the specified topic.
32-
*
33-
* @param topic the target topic to send
34-
* @param body the body data for a message
35-
* @return the created {@code BytesMessage} object
36-
* @throws OMSRuntimeException if the OMS provider fails to create this message due to some internal error.
37-
*/
38-
BytesMessage createTopicBytesMessage(String topic, byte[] body);
39-
4027
/**
4128
* Creates a {@code BytesMessage} object. A {@code BytesMessage} object is used to send a message containing a
4229
* stream of uninterpreted bytes.

‎openmessaging-api/src/main/java/io/openmessaging/MessagingAccessPoint.java

Lines changed: 17 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -22,17 +22,16 @@
2222
import io.openmessaging.consumer.PushConsumer;
2323
import io.openmessaging.consumer.StreamingConsumer;
2424
import io.openmessaging.exception.OMSRuntimeException;
25-
import io.openmessaging.observer.Observer;
2625
import io.openmessaging.producer.Producer;
2726
import java.util.List;
2827

2928
/**
30-
* The {@code MessagingAccessPoint} obtained from {@link MessagingAccessPointFactory} is capable of creating {@code
29+
* The {@code MessagingAccessPoint} obtained from {@link OMS} is capable of creating {@code
3130
* Producer}, {@code Consumer}, {@code ServiceEndPoint}, and so on. <p> For example:
3231
* <pre>
3332
* MessagingAccessPoint messagingAccessPoint = MessagingAccessPointFactory.getMessagingAccessPoint("openmessaging:rocketmq://localhost:10911/namespace");
3433
* Producer producer = messagingAccessPoint.createProducer();
35-
* producer.send(producer.createTopicBytesMessage("HELLO_TOPIC", "HELLO_BODY".getBytes(Charset.forName("UTF-8"))));
34+
* producer.send(producer.createQueueBytesMessage("HELLO_QUEUE", "HELLO_BODY".getBytes(Charset.forName("UTF-8"))));
3635
* </pre>
3736
*
3837
* @version OMS 1.0
@@ -158,30 +157,30 @@ public interface MessagingAccessPoint extends ServiceLifecycle {
158157
ResourceManager getResourceManager();
159158

160159
/**
161-
* Register an observer in an serviceEndPoint object. Whenever serviceEndPoint object publish or bind an service
162-
* object, it will be notified to the list of observer object registered before
160+
* Returns the {@code Producer} list created by the specified {@code MessagingAccessPoint}
163161
*
164-
* @param observer observer event object to an serviceEndPoint object
162+
* @return the producer list
165163
*/
166-
void addObserver(Observer observer);
164+
List<Producer> producers();
167165

168166
/**
169-
* Removes the given observer from the list of observer
170-
* <p>
171-
* If the given observer has not been previously registered (i.e. it was
172-
* never added) then this method call is a no-op. If it had been previously
173-
* added then it will be removed. If it had been added more than once, then
174-
* only the first occurrence will be removed.
167+
* Returns the {@code PushConsumer} list created by the specified {@code MessagingAccessPoint}
175168
*
176-
* @param observer The observer to remove
169+
* @return the push consumer list
177170
*/
178-
void removeObserver(Observer observer);
179-
180-
List<Producer> producers();
181-
182171
List<PushConsumer> pushConsumers();
183172

173+
/**
174+
* Returns the {@code StreamingConsumer} list created by the specified {@code MessagingAccessPoint}
175+
*
176+
* @return the streaming consumer list
177+
*/
184178
List<StreamingConsumer> streamingConsumers();
185179

180+
/**
181+
* Returns the {@code PullConsumer} list created by the specified {@code MessagingAccessPoint}
182+
*
183+
* @return the pull consumer list
184+
*/
186185
List<PullConsumer> pullConsumers();
187186
}

0 commit comments

Comments
 (0)