luciano/amazon-kinesis-client
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge pull request #642 from ychunxue/master

Browse source 
Remove the shutdown event from the queue before executing the shudown…
This commit is contained in:
ychunxue 2019-11-01 15:45:32 -07:00 committed by GitHub
commit 047493aa9e
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
1 changed files with 3 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
amazon-kinesis-client/src/main/java/software/amazon/kinesis/retrieval/fanout/FanOutRecordsPublisher.java
View file

@ -227,7 +227,7 @@ public class FanOutRecordsPublisher implements RecordsPublisher {
// This method is not thread-safe. You need to acquire a lock in the caller in order to execute this.
void executeEventAction(Subscriber<? super RecordsRetrieved> subscriber) {
recordsOrShutdownEvent.apply(recordsEvent -> subscriber.onNext(recordsEvent),
shutdownEvent -> shutdownEvent.getSubscriptionShutdownAction().run());
shutdownEvent -> shutdownEvent.getSubscriptionShutdownAction().run());
}
}
@ -729,7 +729,7 @@ public class FanOutRecordsPublisher implements RecordsPublisher {
executeExceptionOccurred(throwable);
} else {
final SubscriptionShutdownEvent subscriptionShutdownEvent = new SubscriptionShutdownEvent(
() -> executeExceptionOccurred(throwable), "onError", throwable);
() -> {parent.recordsDeliveryQueue.poll(); executeExceptionOccurred(throwable);}, "onError", throwable);
tryEnqueueSubscriptionShutdownEvent(subscriptionShutdownEvent);
}
}
@ -768,7 +768,7 @@ public class FanOutRecordsPublisher implements RecordsPublisher {
executeComplete();
} else {
final SubscriptionShutdownEvent subscriptionShutdownEvent = new SubscriptionShutdownEvent(
() -> executeComplete(), "onComplete");
() -> {parent.recordsDeliveryQueue.poll(); executeComplete();}, "onComplete");
tryEnqueueSubscriptionShutdownEvent(subscriptionShutdownEvent);
}
}