Skip to content

Commit 6b2035e

Browse files
sbSteveKsbSteveK
committed
java x509 sample update
1 parent 98b37d9 commit 6b2035e

File tree

1 file changed

+93
-100
lines changed

1 file changed

+93
-100
lines changed

‎samples/Mqtt/X509PubSub/src/main/java/x509pubsub/X509PubSub.java

Lines changed: 93 additions & 100 deletions
Original file line numberDiff line numberDiff line change
@@ -5,39 +5,21 @@
55

66
package mqtt.x509pubsub;
77

8-
import software.amazon.awssdk.crt.CRT;
9-
import software.amazon.awssdk.crt.CrtResource;
10-
import software.amazon.awssdk.crt.CrtRuntimeException;
11-
import software.amazon.awssdk.crt.io.ClientBootstrap;
12-
import software.amazon.awssdk.crt.mqtt5.*;
13-
import software.amazon.awssdk.crt.mqtt5.Mqtt5ClientOptions.LifecycleEvents;
14-
import software.amazon.awssdk.crt.mqtt5.packets.*;
15-
import software.amazon.awssdk.iot.AwsIotMqtt5ClientBuilder;
16-
17-
import java.util.List;
18-
import java.util.concurrent.CompletableFuture;
19-
20-
// CHECK THESE
218
import java.nio.charset.StandardCharsets;
229
import java.time.Duration;
2310
import java.util.Arrays;
2411
import java.util.UUID;
2512
import java.util.concurrent.CountDownLatch;
2613
import java.util.concurrent.TimeUnit;
2714

15+
import software.amazon.awssdk.crt.CrtResource;
16+
import software.amazon.awssdk.crt.CrtRuntimeException;
17+
import software.amazon.awssdk.crt.mqtt5.*;
18+
import software.amazon.awssdk.crt.mqtt5.packets.*;
19+
import software.amazon.awssdk.iot.AwsIotMqtt5ClientBuilder;
20+
2821
/**
2922
* MQTT5 X509 Sample (mTLS)
30-
*
31-
* Usage (macOS/Linux):
32-
* java -cp target/classes:target/deps/* samples.Mqtt5X509Sample \
33-
* --endpoint <endpoint> \
34-
* --cert </path/certificate.pem> \
35-
* --key </path/private.key> \
36-
* [--ca_file </path/AmazonRootCA1.pem>] \
37-
* [--client-id my-client] \
38-
* [--topic test/topic] \
39-
* [--message "Hello from mqtt5 sample"] \
40-
* [--count 5]
4123
*/
4224
public class X509PubSub {
4325

@@ -110,11 +92,16 @@ public static void main(String[] argv) {
11092
System.out.println("\nStarting MQTT5 X509 PubSub Sample\n");
11193
final int TIMEOUT_SECONDS = 100;
11294

113-
// Latches for flow control of Sample
95+
/*
96+
* Latches for flow control of Sample
97+
*/
11498
CountDownLatch connected = new CountDownLatch(1);
11599
CountDownLatch stopped = new CountDownLatch(1);
116-
CountDownLatch receivedAll = new CountDownLatch(args.count > 0 ? args.count : 1); // if infinite, we won't await fully
100+
CountDownLatch receivedAll = new CountDownLatch(args.count > 0 ? args.count : 1);
117101

102+
/*
103+
* Handle MQTT5 Client lifecycle events
104+
*/
118105
Mqtt5ClientOptions.LifecycleEvents lifecycleEvents = new Mqtt5ClientOptions.LifecycleEvents() {
119106
@Override
120107
public void onAttemptingConnect(Mqtt5Client client, OnAttemptingConnectReturn onAttemptingConnectReturn) {
@@ -152,6 +139,9 @@ public void onStopped(Mqtt5Client client, OnStoppedReturn onStoppedReturn) {
152139
}
153140
};
154141

142+
/*
143+
* Handle Publishes received by the MQTT5 Client
144+
*/
155145
Mqtt5ClientOptions.PublishEvents publishEvents = new Mqtt5ClientOptions.PublishEvents() {
156146
@Override
157147
public void onMessageReceived(Mqtt5Client client, PublishReturn publishReturn) {
@@ -167,102 +157,105 @@ public void onMessageReceived(Mqtt5Client client, PublishReturn publishReturn) {
167157
}
168158
};
169159

160+
Mqtt5Client client;
161+
162+
/**
163+
* Create MQTT5 client using mutual TLS via X509 Certificate and Private Key
164+
*/
165+
System.out.println("==== Creating MQTT5 Client ====\n");
166+
AwsIotMqtt5ClientBuilder builder = AwsIotMqtt5ClientBuilder.newDirectMqttBuilderWithMtlsFromPath(
167+
args.endpoint,args.certPath, args.keyPath);
168+
ConnectPacket.ConnectPacketBuilder connectProperties = new ConnectPacket.ConnectPacketBuilder();
169+
connectProperties.withClientId(args.clientId);
170+
builder.withConnectProperties(connectProperties);
171+
builder.withLifeCycleEvents(lifecycleEvents);
172+
builder.withPublishEvents(publishEvents);
173+
client = builder.build();
174+
builder.close();
175+
176+
System.out.println("==== Starting client ====");
177+
client.start();
170178
try {
171-
Mqtt5Client client;
172-
173-
/**
174-
* Create MQTT5 client using mutual TLS via X509 Certificate and Private Key
175-
*/
176-
System.out.println("==== Creating MQTT5 Client ====\n");
177-
AwsIotMqtt5ClientBuilder builder = AwsIotMqtt5ClientBuilder.newDirectMqttBuilderWithMtlsFromPath(
178-
args.endpoint,args.certPath, args.keyPath);
179-
ConnectPacket.ConnectPacketBuilder connectProperties = new ConnectPacket.ConnectPacketBuilder();
180-
connectProperties.withClientId(args.clientId);
181-
builder.withConnectProperties(connectProperties);
182-
builder.withLifeCycleEvents(lifecycleEvents);
183-
builder.withPublishEvents(publishEvents);
184-
client = builder.build();
185-
builder.close();
186-
187-
188-
System.out.println("==== Starting client ====");
189-
client.start();
190179
if (!connected.await(TIMEOUT_SECONDS, TimeUnit.SECONDS)) {
191180
throw new RuntimeException("Connection timeout");
192181
}
182+
} catch (InterruptedException ex) {
183+
onApplicationFailure(ex);
184+
}
193185

186+
/* Subscribe */
187+
System.out.printf("==== Subscribing to topic '%s' ====%n", args.topic);
188+
SubscribePacket subscribePacket = SubscribePacket.of(args.topic, QOS.AT_LEAST_ONCE);
189+
try {
190+
SubAckPacket subAckPacket = client.subscribe(subscribePacket).get(TIMEOUT_SECONDS, TimeUnit.SECONDS);
191+
System.out.println("SubAck received with reason code:" + subAckPacket.getReasonCodes() + "\n");
192+
} catch (Exception ex) {
193+
onApplicationFailure(ex);
194+
}
194195

195-
/* Subscribe */
196-
System.out.printf("==== Subscribing to topic '%s' ====%n", args.topic);
197-
SubscribePacket subscribePacket = SubscribePacket.of(args.topic, QOS.AT_LEAST_ONCE);
196+
/* Publish */
197+
if (args.count == 0) {
198+
System.out.println("==== Sending messages until program killed ====\n");
199+
} else {
200+
System.out.printf("==== Sending %d message(s) ====%n%n", args.count);
201+
}
202+
int publishCount = 1;
203+
while (args.count == 0 || publishCount <= args.count) {
204+
String payload = args.message + " [" + publishCount + "]";
205+
System.out.printf("Publishing message to topic '%s': %s%n", args.topic, payload);
206+
PublishPacket publishPacket = PublishPacket.of(
207+
args.topic,
208+
QOS.AT_LEAST_ONCE,
209+
payload.getBytes(StandardCharsets.UTF_8));
198210
try {
199-
SubAckPacket subAckPacket = client.subscribe(subscribePacket).get(TIMEOUT_SECONDS, TimeUnit.SECONDS);
200-
System.out.println("SubAck received with reason code:" + subAckPacket.getReasonCodes() + "\n");
211+
PubAckPacket pubAck = client.publish(publishPacket).get(TIMEOUT_SECONDS, TimeUnit.SECONDS).getResultPubAck();
212+
System.out.println("PubAck received with reason: " + pubAck.getReasonCode() + "\n");
201213
} catch (Exception ex) {
202214
onApplicationFailure(ex);
203215
}
204-
205-
/* Publish */
206-
if (args.count == 0) {
207-
System.out.println("==== Sending messages until program killed ====\n");
208-
} else {
209-
System.out.printf("==== Sending %d message(s) ====%n%n", args.count);
210-
}
211-
212-
int publishCount = 1;
213-
while (args.count == 0 || publishCount <= args.count) {
214-
String payload = args.message + " [" + publishCount + "]";
215-
System.out.printf("Publishing message to topic '%s': %s%n", args.topic, payload);
216-
217-
PublishPacket publishPacket = PublishPacket.of(
218-
args.topic,
219-
QOS.AT_LEAST_ONCE,
220-
payload.getBytes(StandardCharsets.UTF_8));
221-
222-
try {
223-
PubAckPacket pubAck = client.publish(publishPacket).get(TIMEOUT_SECONDS, TimeUnit.SECONDS).getResultPubAck();
224-
System.out.println("PubAck received with reason: " + pubAck.getReasonCode() + "\n");
225-
} catch (Exception ex) {
226-
onApplicationFailure(ex);
227-
}
228-
216+
try {
229217
Thread.sleep(Duration.ofMillis(1500).toMillis());
230-
publishCount++;
218+
} catch (InterruptedException ex) {
219+
onApplicationFailure(ex);
231220
}
232-
233-
if (args.count > 0) {
234-
long remaining = receivedAll.getCount();
235-
if (remaining > 0) {
221+
publishCount++;
222+
}
223+
if (args.count > 0) {
224+
long remaining = receivedAll.getCount();
225+
if (remaining > 0) {
226+
try {
236227
receivedAll.await(TIMEOUT_SECONDS, TimeUnit.SECONDS);
228+
} catch (InterruptedException ex) {
229+
onApplicationFailure(ex);
237230
}
238-
long received = (args.count - receivedAll.getCount());
239-
System.out.printf("%d message(s) received.%n%n", received);
240231
}
232+
long received = (args.count - receivedAll.getCount());
233+
System.out.printf("%d message(s) received.%n%n", received);
234+
}
241235

242-
// ---------- Unsubscribe ----------
243-
System.out.printf("==== Unsubscribing from topic '%s' ====%n", args.topic);
244-
UnsubscribePacket unsubscribePacket = UnsubscribePacket.of(args.topic);
245-
try {
246-
UnsubAckPacket unsubAckPacket = client.unsubscribe(unsubscribePacket).get(TIMEOUT_SECONDS, TimeUnit.SECONDS);
247-
System.out.println("UnsubAck received with reason code:" + unsubAckPacket.getReasonCodes() + "\n");
248-
} catch (Exception ex) {
249-
onApplicationFailure(ex);
250-
}
236+
// ---------- Unsubscribe ----------
237+
System.out.printf("==== Unsubscribing from topic '%s' ====%n", args.topic);
238+
UnsubscribePacket unsubscribePacket = UnsubscribePacket.of(args.topic);
239+
try {
240+
UnsubAckPacket unsubAckPacket = client.unsubscribe(unsubscribePacket).get(TIMEOUT_SECONDS, TimeUnit.SECONDS);
241+
System.out.println("UnsubAck received with reason code:" + unsubAckPacket.getReasonCodes() + "\n");
242+
} catch (Exception ex) {
243+
onApplicationFailure(ex);
244+
}
251245

252-
253-
System.out.println("==== Stopping Client ====");
254-
client.stop();
246+
System.out.println("==== Stopping Client ====");
247+
client.stop();
248+
try {
255249
if (!stopped.await(TIMEOUT_SECONDS, TimeUnit.SECONDS)) {
256250
throw new RuntimeException("Stop timeout");
257251
}
258-
System.out.println("==== Client Stopped! ====");
259-
260-
/* Close the client to free memory */
261-
client.close();
262-
263-
} catch (CrtRuntimeException | InterruptedException ex) {
252+
} catch (InterruptedException ex) {
264253
onApplicationFailure(ex);
265254
}
255+
System.out.println("==== Client Stopped! ====");
256+
257+
/* Close the client to free memory */
258+
client.close();
266259

267260
CrtResource.waitForNoResources();
268261
System.out.println("Complete!");

0 commit comments

Comments
 (0)