Skip to content

Commit

Permalink
Pub/Sub: more refactoring (#5044)
Browse files Browse the repository at this point in the history
* Pub/Sub: more refactoring
- Using `Preconditions`
- Adding `onSuccess` and `onFailure` to `OutstandingPublish`

* Fixing a typo

* Fixing formating
  • Loading branch information
sduskis authored May 1, 2019
1 parent 652794a commit 67668c1
Showing 1 changed file with 22 additions and 22 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -192,9 +192,7 @@ public String getTopicNameString() {
* @return the message ID wrapped in a future.
*/
public ApiFuture<String> publish(PubsubMessage message) {
if (shutdown.get()) {
throw new IllegalStateException("Cannot publish on a shut-down publisher.");
}
Preconditions.checkState(!shutdown.get(), "Cannot publish on a shut-down publisher.");

final OutstandingPublish outstandingPublish =
new OutstandingPublish(messageTransform.apply(message));
Expand Down Expand Up @@ -288,23 +286,15 @@ private void publishOutstandingBatch(final OutstandingBatch outstandingBatch) {
public void onSuccess(PublishResponse result) {
try {
if (result.getMessageIdsCount() != outstandingBatch.size()) {
Throwable t =
outstandingBatch.onFailure(
new IllegalStateException(
String.format(
"The publish result count %s does not match "
+ "the expected %s results. Please contact Cloud Pub/Sub support "
+ "if this frequently occurs",
result.getMessageIdsCount(), outstandingBatch.size()));
for (OutstandingPublish oustandingMessage : outstandingBatch.outstandingPublishes) {
oustandingMessage.publishResult.setException(t);
}
return;
}

Iterator<OutstandingPublish> messagesResultsIt =
outstandingBatch.outstandingPublishes.iterator();
for (String messageId : result.getMessageIdsList()) {
messagesResultsIt.next().publishResult.set(messageId);
result.getMessageIdsCount(), outstandingBatch.size())));
} else {
outstandingBatch.onSuccess(result.getMessageIdsList());
}
} finally {
messagesWaiter.incrementPendingMessages(-outstandingBatch.size());
Expand All @@ -314,9 +304,7 @@ public void onSuccess(PublishResponse result) {
@Override
public void onFailure(Throwable t) {
try {
for (OutstandingPublish outstandingPublish : outstandingBatch.outstandingPublishes) {
outstandingPublish.publishResult.setException(t);
}
outstandingBatch.onFailure(t);
} finally {
messagesWaiter.incrementPendingMessages(-outstandingBatch.size());
}
Expand Down Expand Up @@ -350,6 +338,19 @@ private List<PubsubMessage> getMessages() {
}
return results;
}

private void onFailure(Throwable t) {
for (OutstandingPublish outstandingPublish : outstandingPublishes) {
outstandingPublish.publishResult.setException(t);
}
}

private void onSuccess(Iterable<String> results) {
Iterator<OutstandingPublish> messagesResultsIt = outstandingPublishes.iterator();
for (String messageId : results) {
messagesResultsIt.next().publishResult.set(messageId);
}
}
}

private static final class OutstandingPublish {
Expand All @@ -376,10 +377,9 @@ public BatchingSettings getBatchingSettings() {
* should be invoked prior to deleting the {@link Publisher} object in order to ensure that no
* pending messages are lost.
*/
public void shutdown() throws Exception {
if (shutdown.getAndSet(true)) {
throw new IllegalStateException("Cannot shut down a publisher already shut-down.");
}
public void shutdown() {
Preconditions.checkState(
!shutdown.getAndSet(true), "Cannot shut down a publisher already shut-down.");
if (currentAlarmFuture != null && activeAlarm.getAndSet(false)) {
currentAlarmFuture.cancel(false);
}
Expand Down

0 comments on commit 67668c1

Please # to comment.