This repository is no longer active The new Java SDK is here https://github.com/kubemq-io/kubemq-java-v2
The KubeMQ SDK for Java enables Java developers to communicate with KubeMQ server.
Please visit KubeMQ Community for intallation steps.
The SDK implements all communication patterns available through the KubeMQ server:
- Events
- EventStore
- Command
- Query
- Queue
KubeMQ-SDK-Java works with JDK 8+
The recommended way to use the SDK for Java in your project is to consume it from Maven. https://oss.sonatype.org/service/local/repositories/releases/content/io/kubemq/sdk/kubemq-sdk-Java/1.0.6/kubemq-sdk-Java-1.0.6.jar To build with Gradle, add the dependency below to your build.gradle file.
compile group: 'io.kubemq.sdk', name: 'kubemq-sdk-Java', version: '1.0.6'
The only required configuration setting is the KubeMQ server address.
Configuration can be set by using one of the following:
- Environment Variable
- 'Java Property'
Set KubeMQServerAddress
to the KubeMQ Server Address
by passing the -DKubeMQServerAddress= option to the JVM) Within the code
When setting the KubeMQ server address within the code, simply pass the address as a parameter to the various constructors. See exactly how in the code examples in this document.
Javadoc is used for documentation. You can generate HTML locally with the following:
.gradlew javadoc
The examples are standalone projects that showcase the usage of the SDK.
To run the examples, you need to have a running instance of KubeMQ.
You can use the Gradle tasks to run the examples:
.gradlew commandQueryChannel
.gradlew commandQueryInitiator
.gradlew commandQueryResponder
.gradlew commandQueryResponderAsync
.gradlew eventChannel
.gradlew eventSender
.gradlew eventSubscriber
Once you check out the code from GitHub, you can build it using Gradle.
.gradlew build
To run the automated tests for this system execute:
.gradlew test
- Metadata: The metadata allows us to pass additional information with the event. Can be in any form that can be presented as a string, i.e., struct, JSON, XML and many more.
- Body: The actual content of the event. Can be in any form that is serializable into a byte array, i.e., string, struct, JSON, XML, Collection, binary file and many more.
- ClientID: Displayed in logs, tracing, and KubeMQ dashboard(When using Events Store, it must be unique).
- Tags: Set of Key value pair that help categorize the message
- Channel: Represents the endpoint target. One-to-one or one-to-many. Real-Time Multicast.
- Group: Optional parameter when subscribing to a channel. A set of subscribers can define the same group so that only one of the subscribers within the group will receive a specific event. Used mainly for load balancing. Subscribing without the group parameter ensures receiving all the channel messages. (When using Grouping all the programs that are assigned to the group need to have to same channel name)
- Event Store: The Event Store represents a persistence store, should be used when need to store data on a volume.
- Queue: Represents a unique FIFO queue name, used in queue pattern.
- Transaction: Represents an Rpc stream for single message transaction.
- Timestamp - when the message arrived to queue.
- Sequence - the message order in the queue.
- MD5OfBody - An MD5 digest non-URL-encoded message body string.
- ReceiveCount - how many recieved.
- ReRouted - if the message was ReRouted from another point.
- ReRoutedFromQueue - from where the message was ReRouted
- ExpirationAt - Expiration time of the message.
- DelayedTo -if the message was Delayed.
message QueueMessageAttributes {
int64 Timestamp =1;
uint64 Sequence =2;
string MD5OfBody =3;
int32 ReceiveCount =4;
bool ReRouted =5;
string ReRoutedFromQueue =6;
int64 ExpirationAt =7;
int64 DelayedTo =8;
}
A struct that is used to initialize SubscribeToEvents/SubscribeToRequest, the SubscribeRequest contains the following:
- SubscribeType - Mandatory - Enum that represents the subscription type:
- Events - if there is no need for Persistence.
- EventsStore - If you want to receive Events from persistence. See Main concepts.
- Command - Should be used when a response is not needed.
- Query - Should be used when a response is needed.
- ClientID - Mandatory - See Main concepts
- Channel - Mandatory - See Main concepts
- Group - Optional - See Main concepts
- EventsStoreType - Mandatory - set the type event store to subscribe to Main concepts.
KubeMQ supports distributed durable FIFO based queues with the following core features:
- Exactly One Delivery - Only one message guarantee will deliver to the subscriber
- Single and Batch Messages Send and Receive - Single and multiple messages in one call
- RPC and Stream Flow - RPC flow allows an insert and pulls messages in one call. Stream flow allows single message consuming in a transactional way
- Message Policy - Each message can be configured with expiration and delay timers. Also, each message can specify a dead-letter queue for un-processed messages attempts
- Long Polling - Consumers can wait until a message available in the queue to consume
- Peak Messages - Consumers can peek into a queue without removing them from the queue
- Ack All Queue Messages - Any client can mark all the messages in a queue as discarded and will not be available anymore to consume
- Visibility timers - Consumers can pull a message from the queue and set a timer which will cause the message not be visible to other consumers. This timer can be extended as needed.
- Resend Messages - Consumers can send back a message they pulled to a new queue or send a modified message to the same queue for further processing.
Queue queue = new Queue("QueueName", "ClientID", "localhost:50000");
SendMessageResult resSend = queue.SendQueueMessage(new Message()
.setBody(Converter.ToByteArray("some-simple_queue-queue-message"))
.setMetadata("someMeta"));
if (resSend.getIsError()) {
System.out.printf("Message enqueue error, error: %s", resSend.getError());
}
Queue queue = new Queue("QueueName", "ClientID", "localhost:50000");
SendMessageResult resSend = queue
.SendQueueMessage(new Message()
.setBody(Converter.ToByteArray("some-simple_queue-queue-message"))
.setMetadata("someMeta")
.setExpiration(5));
if (resSend.getIsError()) {
System.out.printf("Message enqueue error, error: %s", resSend.getError());
}
Queue queue = new Queue("QueueName", "ClientID", "localhost:50000");
SendMessageResult resSend = queue.SendQueueMessage(new Message()
.setBody(Converter.ToByteArray("some-simple_queue-queue-message"))
.setMetadata("someMeta")
.setDelay(3));
if (resSend.getIsError()) {
System.out.printf("Message enqueue error, error: %s", resSend.getError());
}
Queue queue = new Queue("QueueName", "ClientID", "localhost:50000");
SendMessageResult resSend = queue
.SendQueueMessage(new Message()
.setBody(Converter.ToByteArray("some-simple_queue-queue-message"))
.setMetadata("someMeta")
.setMaxReciveCount(3)
.setMaxReciveQueue("DeadLetterQueue"));
if (resSend.getIsError()) {
System.out.printf("Message enqueue error, error: %s", resSend.getError());
}
Queue queue = new Queue("QueueName", "ClientID", "localhost:50000");
Collection<Message> batch = new ArrayList<Message>();
for (int i = 0; i < 10; i++) {
batch.add(new Message()
.setBody(Converter.ToByteArray("Batch Message " + i)));
}
SendBatchMessageResult resBatch = queue.SendQueueMessagesBatch(batch);
if (resBatch.getHaveErrors()) {
System.out.print("Message sent batch has errors");
}
for (SendMessageResult resSend : resBatch.getResults()) {
if (resSend.getIsError()) {
System.out.printf("Message enqueue error, error: %s", resSend.getError());
} else {
System.out.printf("Send to Queue Result: MessageID: %s, Sent At:%s", resSend.getMessageID(),
Converter.FromUnixTime(resSend.getSentAt()).toString());
}
}
Queue queue = new Queue("QueueName", "ClientID", "localhost:50000");
ReceiveMessagesResponse resRec = queue.ReceiveQueueMessages(10, 1);
if (resRec.getIsError()) {
System.out.printf("Message dequeue error, error: %s", resRec.getError());
return;
}
System.out.printf("Received Messages %s:", resRec.getMessagesReceived());
for (Message msg : resRec.getMessages()) {
System.out.printf("MessageID: %s, Body:%s", msg.getMessageID(), Converter.FromByteArray(msg.getBody()));
}
Queue queue = new Queue("QueueName", "ClientID", "localhost:50000");
ReceiveMessagesResponse resPek = queue.PeekQueueMessage(10, 1);
if (resPek.getIsError()) {
System.out.printf("Message dequeue error, error: %s", resPek.getError());
return;
}
System.out.printf("Received Messages: %s", resPek.getMessagesReceived());
for (Message msg : resPek.getMessages()) {
System.out.printf("MessageID: %s, Body: %s", msg.getMessageID(), Converter.FromByteArray(msg.getBody()));
}
Queue queue = new Queue("QueueName", "ClientID", "localhost:50000");
AckAllMessagesResponse resAck = queue.AckAllQueueMessages();
if (resAck.getIsError()) {
System.out.printf("AckAllQueueMessagesResponse error, error: %s", resAck.getError());
return;
}
System.out.printf("Ack All Messages: %d completed", resAck.getAffectedMessages());
Queue queue = new Queue("QueueName", "ClientID", "localhost:50000");
Transaction tran = queue.CreateTransaction();
TransactionMessagesResponse resRec = tran.Receive(10, 10);
if (resRec.getIsError()) {
System.out.printf("Message dequeue error, error: %s", resRec.getError());
return;
}
System.out.printf("MessageID: %d, Body: %s", resRec.getMessage().getMessageID(),
Converter.FromByteArray(resRec.getMessage().getBody()));
System.out.println("Doing some work.....");
Thread.sleep(1000);
System.out.println("Done, ack the message");
TransactionMessagesResponse resAck = tran.AckMessage();
if (resAck.getIsError()) {
System.out.printf("Ack message error: %s", resAck.getError());
}
Queue queue = new Queue("QueueName", "ClientID", "localhost:50000");
Transaction tran = queue.CreateTransaction();
TransactionMessagesResponse resRec = tran.Receive(10, 10);
if (resRec.getIsError()) {
System.out.printf("Message dequeue error, error: %s", resRec.getError());
return;
}
System.out.printf("MessageID: %d, Body: %s", resRec.getMessage().getMessageID(),
Converter.FromByteArray(resRec.getMessage().getBody()));
System.out.println("Reject message");
TransactionMessagesResponse resRej = tran.RejectMessage();
if (resRej.getIsError()) {
System.out.printf("Message dequeue error, error: %s", resRej.getError());
return;
}
Queue queue = new Queue("QueueName", "ClientID", "localhost:50000");
Transaction tran = queue.CreateTransaction();
TransactionMessagesResponse resRec = tran.Receive(5, 10);
if (resRec.getIsError()) {
System.out.printf("Message dequeue error, error: %s", resRec.getError());
return;
}
System.out.printf("MessageID: %d, Body: %s", resRec.getMessage().getMessageID(),
Converter.FromByteArray(resRec.getMessage().getBody()));
System.out.println("work for 1 seconds");
Thread.sleep(1000);
System.out.println("Need more time to process, extend visibility for more 3 seconds");
TransactionMessagesResponse resExt = tran.ExtendVisibility(3);
if (resExt.getIsError()) {
System.out.printf("Message dequeue error, error: %s", resExt.getError());
return;
}
System.out.println("Approved. work for 2.5 seconds");
Thread.sleep(2500);
System.out.println("Work done... ack the message");
TransactionMessagesResponse resAck = tran.AckMessage();
if (resAck.getIsError()) {
System.out.printf("Ack message error: %s", resAck.getError());
}
System.out.println("Ack done");
Queue queue = new Queue("QueueName", "ClientID", "localhost:50000");
Transaction tran = queue.CreateTransaction();
TransactionMessagesResponse resRec = tran.Receive(500, 10);
if (resRec.getIsError()) {
System.out.printf("Message dequeue error, error: %s", resRec.getError());
return;
}
System.out.printf("MessageID: %d, Body:%s", resRec.getMessage().getMessageID(),
Converter.FromByteArray(resRec.getMessage().getBody()));
TransactionMessagesResponse resMod = tran
.Modify(resRec.getMessage().setQueue("receiverB").setMetadata("new meatdata"));
if (resMod.getIsError()) {
System.out.printf("Message Modify error, error::%s", resMod.getError());
return;
}
Queue queue = new Queue("QueueName", "ClientID", "localhost:50000");
Transaction tran = queue.CreateTransaction();
TransactionMessagesResponse resRec = tran.Receive(5, 10);
if (resRec.getIsError()) {
System.out.printf("Message dequeue error, error: %s", resRec.getError());
return;
}
System.out.printf("MessageID: %d, Body:%s", resRec.getMessage().getMessageID(),
Converter.FromByteArray(resRec.getMessage().getBody()));
System.out.println("Resend to new queue");
TransactionMessagesResponse resResend = tran.ReSend("new-queue");
if (resResend.getIsError()) {
System.out.printf("Message dequeue error, error: %s", resResend.getError());
return;
}
System.out.println("Done");
String ChannelName = "testing_event_channel", ClientID = "hello-world-sender",
KubeMQServerAddress = "localhost:50000";
io.kubemq.sdk.event.Channel channel = new io.kubemq.sdk.event.Channel(ChannelName, ClientID, false,
KubeMQServerAddress);
Event event = new Event();
event.setBody(Converter.ToByteArray("hello kubemq - sending single event"));
Result result;
try {
result = channel.SendEvent(event);
if (!result.isSent()) {
System.out.println("Could not send single message");
return;
}
} catch (ServerAddressNotSuppliedException e) {
System.out.printf("Could not send single message: %s", e.getMessage());
e.printStackTrace();
}
String ChannelName = "testing_event_channel", ClientID = "hello-world-sender",
KubeMQServerAddress = "localhost:50000";
io.kubemq.sdk.event.Channel channel = new io.kubemq.sdk.event.Channel(ChannelName, ClientID, false,
KubeMQServerAddress);
Event event = new Event();
event.setBody(Converter.ToByteArray("hello kubemq - sending single event"));
StreamObserver<Result> streamResObserver = new StreamObserver<Result>() {
@Override
public void onNext(Result value) {
System.out.printf("Sent event: %s", value.getEventId());
}
@Override
public void onError(Throwable t) {
System.out.printf("Could not send single message");
}
@Override
public void onCompleted() {
}
};
StreamObserver<Event> stream = channel.StreamEvent(streamResObserver);
stream.onNext(event);
String ChannelName = "testing_event_channel", ClientID = "hello-world-subscriber",
KubeMQServerAddress = "localhost:50000";
Subscriber subscriber = new Subscriber(KubeMQServerAddress);
SubscribeRequest subscribeRequest = new SubscribeRequest();
subscribeRequest.setChannel(ChannelName);
subscribeRequest.setClientID(ClientID);
subscribeRequest.setSubscribeType(SubscribeType.Events);
StreamObserver<EventReceive> streamObserver = new StreamObserver<EventReceive>() {
@Override
public void onNext(EventReceive value) {
try {
System.out.printf("Event Received: EventID: %d, Channel: %s, Metadata: %s, Body: %s",
value.getEventId(), value.getChannel(), value.getMetadata(),
Converter.FromByteArray(value.getBody()));
} catch (ClassNotFoundException e) {
System.out.printf("ClassNotFoundException: %s", e.getMessage());
e.printStackTrace();
} catch (IOException e) {
System.out.printf("IOException: %s", e.getMessage());
e.printStackTrace();
}
}
@Override
public void onError(Throwable t) {
System.out.printf("Event Received Error: %s", t.toString());
}
@Override
public void onCompleted() {
}
};
subscriber.SubscribeToEvents(subscribeRequest, streamObserver);
KubeMQ supports 6 types of subscriptions:
-
StartFromNewEvents - start event store subscription with only new events
-
StartFromFirstEvent - replay all the stored events from the first available sequence and continue stream new events from this point
-
StartFromLastEvent - replay the last event and continue stream new events from this point
-
StartFromSequence - replay events from specific event sequence number and continue stream new events from this point
-
StartFromTime - replay events from specific time continue stream new events from this point
-
StartFromTimeDelta - replay events from specific current time - delta duration in seconds, continue stream new events from this point
String ChannelName = "testing_event_channel", ClientID = "hello-world-sender",
KubeMQServerAddress = "localhost:50000";
io.kubemq.sdk.event.Channel channel = new io.kubemq.sdk.event.Channel(ChannelName, ClientID, false,
KubeMQServerAddress);
for (int i = 0; i < 10; i++) {
Event event = new Event();
event.setBody(Converter.ToByteArray("hello kubemq - sending single event"));
event.setEventId("event-Store-" + i);
try {
channel.SendEvent(event);
} catch (SSLException e) {
System.out.printf("SSLException: %s", e.getMessage());
e.printStackTrace();
} catch (ServerAddressNotSuppliedException e) {
System.out.printf("ServerAddressNotSuppliedException: %s", e.getMessage());
e.printStackTrace();
}
}
String ChannelName = "testing_event_channel", ClientID = "hello-world-sender",
KubeMQServerAddress = "localhost:50000";
io.kubemq.sdk.event.Channel channel = new io.kubemq.sdk.event.Channel(ChannelName, ClientID, false,
KubeMQServerAddress);
StreamObserver<Result> streamResObserver = new StreamObserver<Result>() {
@Override
public void onNext(Result value) {
System.out.printf("Stream event: %s", value.getEventId());
}
@Override
public void onError(Throwable t) {
System.out.printf("Could not send single message");
}
@Override
public void onCompleted() {
}
};
StreamObserver<Event> stream = channel.StreamEvent(streamResObserver);
for (int i = 0; i < 10; i++) {
Event event = new Event();
event.setBody(Converter.ToByteArray("hello kubemq - sending single event"));
event.setEventId("event-Store-" + i);
stream.onNext(event);
}
String ChannelName = "testing_event_channel", ClientID = "hello-world-subscriber",
KubeMQServerAddress = "localhost:50000";
Subscriber subscriber = new Subscriber(KubeMQServerAddress);
SubscribeRequest subscribeRequest = new SubscribeRequest();
subscribeRequest.setChannel(ChannelName);
subscribeRequest.setClientID(ClientID);
subscribeRequest.setSubscribeType(SubscribeType.EventsStore);
subscribeRequest.setEventsStoreType(EventsStoreType.StartAtSequence);
StreamObserver<EventReceive> streamObserver = new StreamObserver<EventReceive>() {
@Override
public void onNext(EventReceive value) {
try {
System.out.printf("Event Received: EventID: %s, Channel: %s, Metadata: %s, Body: %s",
value.getEventId(), value.getChannel(), value.getMetadata(),
Converter.FromByteArray(value.getBody()));
} catch (ClassNotFoundException e) {
System.out.printf("ClassNotFoundException: %s", e.getMessage());
e.printStackTrace();
} catch (IOException e) {
System.out.printf("IOException: %s", e.getMessage());
e.printStackTrace();
}
}
@Override
public void onError(Throwable t) {
System.out.printf("onError: %s", t.getMessage());
}
@Override
public void onCompleted() {
}
};
subscriber.SubscribeToEvents(subscribeRequest, streamObserver);
Commands implement synchronous messaging pattern which the sender send a request and wait for a specific amount of time to get a response.
The response can be successful or not. This is the responsibility of the responder to return with the result of the command within the time the sender set in the request
String ChannelName = "testing_Command_channel", ClientID = "hello-world-sender",
KubeMQServerAddress = "localhost:50000";
Responder.RequestResponseObserver HandleIncomingRequests;
Responder responder = new Responder(KubeMQServerAddress);
HandleIncomingRequests = request -> {
Response response = new Response(request);
response.setCacheHit(false);
response.setError("None");
response.setClientID(ClientID);
response.setBody("OK".getBytes());
response.setExecuted(true);
response.setMetadata("OK");
response.setTimestamp(LocalDateTime.now());
return response;
};
SubscribeRequest subscribeRequest = new SubscribeRequest();
subscribeRequest.setChannel(ChannelName);
subscribeRequest.setClientID(ClientID);
subscribeRequest.setSubscribeType(SubscribeType.Commands);
new Thread() {
public void run() {
try {
responder.SubscribeToRequests(subscribeRequest, HandleIncomingRequests);
} catch (SSLException e) {
System.out.printf("SSLException:%s", e.getMessage());
e.printStackTrace();
} catch (ServerAddressNotSuppliedException e) {
System.out.printf("ServerAddressNotSuppliedException:%s", e.getMessage());
e.printStackTrace();
}
}
}.start();
String ChannelName = "testing_Command_channel", ClientID = "hello-world-sender",
KubeMQServerAddress = "localhost:50000";
io.kubemq.sdk.commandquery.ChannelParameters channelParameters = new io.kubemq.sdk.commandquery.ChannelParameters();
channelParameters.setChannelName(ChannelName);
channelParameters.setClientID(ClientID);
channelParameters.setKubeMQAddress(KubeMQServerAddress);
channelParameters.setRequestType(RequestType.Command);
channelParameters.setTimeout(10000);
io.kubemq.sdk.commandquery.Channel channel = new io.kubemq.sdk.commandquery.Channel(channelParameters);
Request request = new Request();
request.setBody(Converter.ToByteArray("hello kubemq - sending a command, please reply"));
Response result = channel.SendRequest(request);
if (!result.isExecuted()) {
System.out.printf("Response error: %s", result.getError());
return;
}
System.out.printf("Response Received: %s, ExecutedAt: %d", result.getRequestID(), result.getTimestamp());
String ChannelName = "testing_Command_channel", ClientID = "hello-world-sender",
KubeMQServerAddress = "localhost:50000";
io.kubemq.sdk.commandquery.ChannelParameters channelParameters = new io.kubemq.sdk.commandquery.ChannelParameters();
channelParameters.setChannelName(ChannelName);
channelParameters.setClientID(ClientID);
channelParameters.setKubeMQAddress(KubeMQServerAddress);
channelParameters.setRequestType(RequestType.Command);
channelParameters.setTimeout(1000);
io.kubemq.sdk.commandquery.Channel channel = new io.kubemq.sdk.commandquery.Channel(channelParameters);
Request request = new Request();
request.setBody(Converter.ToByteArray("hello kubemq - sending a command, please reply"));
StreamObserver<Response> response = new StreamObserver<Response>() {
@Override
public void onNext(Response value) {
if (!value.isExecuted()) {
System.out.printf("Response error: %s", value.getError());
}
}
@Override
public void onError(Throwable t) {
System.out.printf("RPC Error: %s", t.getMessage());
}
@Override
public void onCompleted() {
}
};
channel.SendRequestAsync(request, response);
Queries implement synchronous messaging pattern which the sender send a request and wait for a specific amount of time to get a response.
The response must include metadata or body together with an indication of successful or not operation. This is the responsibility of the responder to return with the result of the query within the time the sender set in the request.
String ChannelName = "testing_Command_channel", ClientID = "hello-world-sender",
KubeMQServerAddress = "localhost:50000";
Responder.RequestResponseObserver HandleIncomingRequests;
Responder responder = new Responder(KubeMQServerAddress);
HandleIncomingRequests = request -> {
Response response = new Response(request);
response.setCacheHit(false);
response.setError("None");
response.setClientID(ClientID);
response.setBody("got your query, you are good to goo".getBytes());
response.setExecuted(true);
response.setMetadata("this is a response");
response.setTimestamp(LocalDateTime.now());
return response;
};
SubscribeRequest subscribeRequest = new SubscribeRequest();
subscribeRequest.setChannel(ChannelName);
subscribeRequest.setClientID(ClientID);
subscribeRequest.setSubscribeType(SubscribeType.Queries);
new Thread() {
public void run() {
try {
responder.SubscribeToRequests(subscribeRequest, HandleIncomingRequests);
} catch (SSLException e) {
System.out.printf("SSLException: %s", e.getMessage());
e.printStackTrace();
} catch (ServerAddressNotSuppliedException e) {
System.out.printf("ServerAddressNotSuppliedException: %s", e.getMessage());
e.printStackTrace();
}
}
}.start();
String ChannelName = "testing_Command_channel", ClientID = "hello-world-sender",
KubeMQServerAddress = "localhost:50000";
io.kubemq.sdk.commandquery.ChannelParameters channelParameters = new io.kubemq.sdk.commandquery.ChannelParameters();
channelParameters.setChannelName(ChannelName);
channelParameters.setClientID(ClientID);
channelParameters.setKubeMQAddress(KubeMQServerAddress);
channelParameters.setRequestType(RequestType.Query);
channelParameters.setTimeout(1000);
io.kubemq.sdk.commandquery.Channel channel = new io.kubemq.sdk.commandquery.Channel(channelParameters);
Request request = new Request();
request.setBody(Converter.ToByteArray("hello kubemq - sending a query, please reply"));
Response result = channel.SendRequest(request);
if (!result.isExecuted()) {
System.out.printf("Response error: %s", result.getError());
return;
}
System.out.printf("Response Received: %s, ExecutedAt: %d", result.getRequestID(), result.getTimestamp());
String ChannelName = "testing_Command_channel", ClientID = "hello-world-sender",
KubeMQServerAddress = "localhost:50000";
io.kubemq.sdk.commandquery.ChannelParameters channelParameters = new io.kubemq.sdk.commandquery.ChannelParameters();
channelParameters.setChannelName(ChannelName);
channelParameters.setClientID(ClientID);
channelParameters.setKubeMQAddress(KubeMQServerAddress);
channelParameters.setRequestType(RequestType.Query);
channelParameters.setTimeout(1000);
io.kubemq.sdk.commandquery.Channel channel = new io.kubemq.sdk.commandquery.Channel(channelParameters);
Request request = new Request();
request.setBody(Converter.ToByteArray("hello kubemq - sending a query, please reply"));
StreamObserver<Response> response = new StreamObserver<Response>() {
@Override
public void onNext(Response value) {
if (!value.isExecuted()) {
System.out.printf("Response error: %s", value.getError());
System.out.printf("Response Received: %s, ExecutedAt %d", value.getRequestID(),
value.getTimestamp());
}
}
@Override
public void onError(Throwable t) {
System.out.printf("onError: %s", t.getMessage());
}
@Override
public void onCompleted() {
}
};
channel.SendRequestAsync(request, response);