Skip to content

Commit b9fcb30

Browse files
Justin Boswelljustinboswell
Justin Boswell
authored and
justinboswell
committed
Generated QoS into each service API, updated samples to use qos
1 parent 47714d8 commit b9fcb30

4 files changed

Lines changed: 239 additions & 103 deletions

File tree

‎samples/Jobs/src/main/java/jobs/JobsSample.java‎

Lines changed: 47 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323
import software.amazon.awssdk.crt.mqtt.MqttClient;
2424
import software.amazon.awssdk.crt.mqtt.MqttConnection;
2525
import software.amazon.awssdk.crt.mqtt.MqttConnectionEvents;
26+
import software.amazon.awssdk.crt.mqtt.QualityOfService;
2627
import software.amazon.awssdk.iot.iotjobs.IotJobsClient;
2728
import software.amazon.awssdk.iot.iotjobs.model.DescribeJobExecutionRequest;
2829
import software.amazon.awssdk.iot.iotjobs.model.DescribeJobExecutionResponse;
@@ -208,7 +209,9 @@ public void onConnectionResumed(boolean sessionPresent) {
208209
GetPendingJobExecutionsSubscriptionRequest subscriptionRequest = new GetPendingJobExecutionsSubscriptionRequest();
209210
subscriptionRequest.thingName = "crt-test";
210211
CompletableFuture<Integer> subscribed = jobs.SubscribeToGetPendingJobExecutionsAccepted(
211-
subscriptionRequest, JobsSample::onGetPendingJobExecutionsAccepted)
212+
subscriptionRequest,
213+
QualityOfService.AT_LEAST_ONCE,
214+
JobsSample::onGetPendingJobExecutionsAccepted)
212215
.exceptionally((ex) -> {
213216
System.out.println("Failed to subscribe to GetPendingJobExecutions: " + ex.toString());
214217
return null;
@@ -218,13 +221,18 @@ public void onConnectionResumed(boolean sessionPresent) {
218221

219222
gotResponse = new CompletableFuture<>();
220223

221-
subscribed = jobs.SubscribeToGetPendingJobExecutionsRejected(subscriptionRequest, JobsSample::onRejectedError);
224+
subscribed = jobs.SubscribeToGetPendingJobExecutionsRejected(
225+
subscriptionRequest,
226+
QualityOfService.AT_LEAST_ONCE,
227+
JobsSample::onRejectedError);
222228
subscribed.get();
223229
System.out.println("Subscribed to GetPendingJobExecutionsRejected");
224230

225231
GetPendingJobExecutionsRequest publishRequest = new GetPendingJobExecutionsRequest();
226232
publishRequest.thingName = thingName;
227-
CompletableFuture<Integer> published = jobs.PublishGetPendingJobExecutions(publishRequest)
233+
CompletableFuture<Integer> published = jobs.PublishGetPendingJobExecutions(
234+
publishRequest,
235+
QualityOfService.AT_LEAST_ONCE)
228236
.exceptionally((ex) -> {
229237
System.out.println("Exception occurred during publish: " + ex.toString());
230238
gotResponse.complete(null);
@@ -245,15 +253,21 @@ public void onConnectionResumed(boolean sessionPresent) {
245253
DescribeJobExecutionSubscriptionRequest subscriptionRequest = new DescribeJobExecutionSubscriptionRequest();
246254
subscriptionRequest.thingName = thingName;
247255
subscriptionRequest.jobId = jobId;
248-
jobs.SubscribeToDescribeJobExecutionAccepted(subscriptionRequest, JobsSample::onDescribeJobExecutionAccepted);
249-
jobs.SubscribeToDescribeJobExecutionRejected(subscriptionRequest, JobsSample::onRejectedError);
256+
jobs.SubscribeToDescribeJobExecutionAccepted(
257+
subscriptionRequest,
258+
QualityOfService.AT_LEAST_ONCE,
259+
JobsSample::onDescribeJobExecutionAccepted);
260+
jobs.SubscribeToDescribeJobExecutionRejected(
261+
subscriptionRequest,
262+
QualityOfService.AT_LEAST_ONCE,
263+
JobsSample::onRejectedError);
250264

251265
DescribeJobExecutionRequest publishRequest = new DescribeJobExecutionRequest();
252266
publishRequest.thingName = thingName;
253267
publishRequest.jobId = jobId;
254268
publishRequest.includeJobDocument = true;
255269
publishRequest.executionNumber = 1L;
256-
jobs.PublishDescribeJobExecution(publishRequest);
270+
jobs.PublishDescribeJobExecution(publishRequest, QualityOfService.AT_LEAST_ONCE);
257271
gotResponse.get();
258272
}
259273

@@ -265,13 +279,19 @@ public void onConnectionResumed(boolean sessionPresent) {
265279
StartNextPendingJobExecutionSubscriptionRequest subscriptionRequest = new StartNextPendingJobExecutionSubscriptionRequest();
266280
subscriptionRequest.thingName = thingName;
267281

268-
jobs.SubscribeToStartNextPendingJobExecutionAccepted(subscriptionRequest, JobsSample::onStartNextPendingJobExecutionAccepted);
269-
jobs.SubscribeToStartNextPendingJobExecutionRejected(subscriptionRequest, JobsSample::onRejectedError);
282+
jobs.SubscribeToStartNextPendingJobExecutionAccepted(
283+
subscriptionRequest,
284+
QualityOfService.AT_LEAST_ONCE,
285+
JobsSample::onStartNextPendingJobExecutionAccepted);
286+
jobs.SubscribeToStartNextPendingJobExecutionRejected(
287+
subscriptionRequest,
288+
QualityOfService.AT_LEAST_ONCE,
289+
JobsSample::onRejectedError);
270290

271291
StartNextPendingJobExecutionRequest publishRequest = new StartNextPendingJobExecutionRequest();
272292
publishRequest.thingName = thingName;
273293
publishRequest.stepTimeoutInMinutes = 15L;
274-
jobs.PublishStartNextPendingJobExecution(publishRequest);
294+
jobs.PublishStartNextPendingJobExecution(publishRequest, QualityOfService.AT_LEAST_ONCE);
275295

276296
gotResponse.get();
277297
}
@@ -283,19 +303,25 @@ public void onConnectionResumed(boolean sessionPresent) {
283303
UpdateJobExecutionSubscriptionRequest subscriptionRequest = new UpdateJobExecutionSubscriptionRequest();
284304
subscriptionRequest.thingName = thingName;
285305
subscriptionRequest.jobId = currentJobId;
286-
jobs.SubscribeToUpdateJobExecutionAccepted(subscriptionRequest, (response) -> {
306+
jobs.SubscribeToUpdateJobExecutionAccepted(
307+
subscriptionRequest,
308+
QualityOfService.AT_LEAST_ONCE,
309+
(response) -> {
287310
System.out.println("Marked job " + currentJobId + " IN_PROGRESS");
288311
gotResponse.complete(null);
289312
});
290-
jobs.SubscribeToUpdateJobExecutionRejected(subscriptionRequest, JobsSample::onRejectedError);
313+
jobs.SubscribeToUpdateJobExecutionRejected(
314+
subscriptionRequest,
315+
QualityOfService.AT_LEAST_ONCE,
316+
JobsSample::onRejectedError);
291317

292318
UpdateJobExecutionRequest publishRequest = new UpdateJobExecutionRequest();
293319
publishRequest.thingName = thingName;
294320
publishRequest.jobId = currentJobId;
295321
publishRequest.executionNumber = currentExecutionNumber;
296322
publishRequest.status = JobStatus.IN_PROGRESS;
297323
publishRequest.expectedVersion = currentVersionNumber++;
298-
jobs.PublishUpdateJobExecution(publishRequest);
324+
jobs.PublishUpdateJobExecution(publishRequest, QualityOfService.AT_LEAST_ONCE);
299325

300326
gotResponse.get();
301327
}
@@ -310,19 +336,25 @@ public void onConnectionResumed(boolean sessionPresent) {
310336
UpdateJobExecutionSubscriptionRequest subscriptionRequest = new UpdateJobExecutionSubscriptionRequest();
311337
subscriptionRequest.thingName = thingName;
312338
subscriptionRequest.jobId = currentJobId;
313-
jobs.SubscribeToUpdateJobExecutionAccepted(subscriptionRequest, (response) -> {
339+
jobs.SubscribeToUpdateJobExecutionAccepted(
340+
subscriptionRequest,
341+
QualityOfService.AT_LEAST_ONCE,
342+
(response) -> {
314343
System.out.println("Marked job " + currentJobId + " SUCCEEDED");
315344
gotResponse.complete(null);
316345
});
317-
jobs.SubscribeToUpdateJobExecutionRejected(subscriptionRequest, JobsSample::onRejectedError);
346+
jobs.SubscribeToUpdateJobExecutionRejected(
347+
subscriptionRequest,
348+
QualityOfService.AT_LEAST_ONCE,
349+
JobsSample::onRejectedError);
318350

319351
UpdateJobExecutionRequest publishRequest = new UpdateJobExecutionRequest();
320352
publishRequest.thingName = thingName;
321353
publishRequest.jobId = currentJobId;
322354
publishRequest.executionNumber = currentExecutionNumber;
323355
publishRequest.status = JobStatus.SUCCEEDED;
324356
publishRequest.expectedVersion = currentVersionNumber++;
325-
jobs.PublishUpdateJobExecution(publishRequest);
357+
jobs.PublishUpdateJobExecution(publishRequest, QualityOfService.AT_LEAST_ONCE);
326358

327359
gotResponse.get();
328360
}

‎samples/Shadow/src/main/java/jobs/ShadowSample.java‎

Lines changed: 25 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323
import software.amazon.awssdk.crt.mqtt.MqttClient;
2424
import software.amazon.awssdk.crt.mqtt.MqttConnection;
2525
import software.amazon.awssdk.crt.mqtt.MqttConnectionEvents;
26+
import software.amazon.awssdk.crt.mqtt.QualityOfService;
2627
import software.amazon.awssdk.iot.iotshadow.IotShadowClient;
2728
import software.amazon.awssdk.iot.iotshadow.model.ErrorResponse;
2829
import software.amazon.awssdk.iot.iotshadow.model.GetShadowRequest;
@@ -201,7 +202,7 @@ static CompletableFuture<Void> changeShadowValue(String value) {
201202
}};
202203

203204
// Publish the request
204-
return shadow.PublishUpdateShadow(request).thenRun(() -> {
205+
return shadow.PublishUpdateShadow(request, QualityOfService.AT_LEAST_ONCE).thenRun(() -> {
205206
System.out.println("Update request published");
206207
}).exceptionally((ex) -> {
207208
System.out.println("Update request failed: " + ex.getMessage());
@@ -255,26 +256,41 @@ public void onConnectionResumed(boolean sessionPresent) {
255256
ShadowDeltaUpdatedSubscriptionRequest requestShadowDeltaUpdated = new ShadowDeltaUpdatedSubscriptionRequest();
256257
requestShadowDeltaUpdated.thingName = thingName;
257258
CompletableFuture<Integer> subscribedToDeltas =
258-
shadow.SubscribeToShadowDeltaUpdatedEvents(requestShadowDeltaUpdated, ShadowSample::onShadowDeltaUpdated);
259+
shadow.SubscribeToShadowDeltaUpdatedEvents(
260+
requestShadowDeltaUpdated,
261+
QualityOfService.AT_LEAST_ONCE,
262+
ShadowSample::onShadowDeltaUpdated);
259263
subscribedToDeltas.get();
260264

261265
System.out.println("Subscribing to update respones...");
262266
UpdateShadowSubscriptionRequest requestUpdateShadow = new UpdateShadowSubscriptionRequest();
263267
requestUpdateShadow.thingName = thingName;
264268
CompletableFuture<Integer> subscribedToUpdateAccepted =
265-
shadow.SubscribeToUpdateShadowAccepted(requestUpdateShadow, ShadowSample::onUpdateShadowAccepted);
269+
shadow.SubscribeToUpdateShadowAccepted(
270+
requestUpdateShadow,
271+
QualityOfService.AT_LEAST_ONCE,
272+
ShadowSample::onUpdateShadowAccepted);
266273
CompletableFuture<Integer> subscribedToUpdateRejected =
267-
shadow.SubscribeToUpdateShadowRejected(requestUpdateShadow, ShadowSample::onUpdateShadowRejected);
274+
shadow.SubscribeToUpdateShadowRejected(
275+
requestUpdateShadow,
276+
QualityOfService.AT_LEAST_ONCE,
277+
ShadowSample::onUpdateShadowRejected);
268278
subscribedToUpdateAccepted.get();
269279
subscribedToUpdateRejected.get();
270280

271281
System.out.println("Subscribing to get responses...");
272282
GetShadowSubscriptionRequest requestGetShadow = new GetShadowSubscriptionRequest();
273283
requestGetShadow.thingName = thingName;
274284
CompletableFuture<Integer> subscribedToGetShadowAccepted =
275-
shadow.SubscribeToGetShadowAccepted(requestGetShadow, ShadowSample::onGetShadowAccepted);
285+
shadow.SubscribeToGetShadowAccepted(
286+
requestGetShadow,
287+
QualityOfService.AT_LEAST_ONCE,
288+
ShadowSample::onGetShadowAccepted);
276289
CompletableFuture<Integer> subscribedToGetShadowRejected =
277-
shadow.SubscribeToGetShadowRejected(requestGetShadow, ShadowSample::onGetShadowRejected);
290+
shadow.SubscribeToGetShadowRejected(
291+
requestGetShadow,
292+
QualityOfService.AT_LEAST_ONCE,
293+
ShadowSample::onGetShadowRejected);
278294
subscribedToGetShadowAccepted.get();
279295
subscribedToGetShadowRejected.get();
280296

@@ -283,7 +299,9 @@ public void onConnectionResumed(boolean sessionPresent) {
283299
System.out.println("Requesting current shadow state...");
284300
GetShadowRequest getShadowRequest = new GetShadowRequest();
285301
getShadowRequest.thingName = thingName;
286-
CompletableFuture<Integer> publishedGetShadow = shadow.PublishGetShadow(getShadowRequest);
302+
CompletableFuture<Integer> publishedGetShadow = shadow.PublishGetShadow(
303+
getShadowRequest,
304+
QualityOfService.AT_LEAST_ONCE);
287305
publishedGetShadow.get();
288306
gotResponse.get();
289307

0 commit comments

Comments
 (0)