-
Notifications
You must be signed in to change notification settings - Fork 5
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: add onConnectionLost and onConnectionRestored callback functions (
#342)
- Loading branch information
1 parent
4460c67
commit 51ca3b4
Showing
7 changed files
with
256 additions
and
8 deletions.
There are no files selected for viewing
34 changes: 34 additions & 0 deletions
34
momento-sdk/src/main/java/momento/sdk/IScsTopicConnection.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,34 @@ | ||
| package momento.sdk; | ||
|
|
||
| import grpc.cache_client.pubsub._SubscriptionItem; | ||
| import grpc.cache_client.pubsub._SubscriptionRequest; | ||
|
|
||
| /** Represents a connection to an ScsTopic for subscribing to events. */ | ||
| interface IScsTopicConnection { | ||
|
|
||
| /** | ||
| * Closes the connection. | ||
| * | ||
| * <p>Note: This method is intended for testing purposes and should never be called from outside | ||
| * of tests. | ||
| */ | ||
| void close(); | ||
|
|
||
| /** | ||
| * Opens the connection. | ||
| * | ||
| * <p>Note: This method is intended for testing purposes and should never be called from outside | ||
| * of tests. | ||
| */ | ||
| void open(); | ||
|
|
||
| /** | ||
| * Subscribes to a specific topic using the provided subscription request and observer. | ||
| * | ||
| * @param subscriptionRequest The subscription request containing details about the subscription. | ||
| * @param subscription The observer to handle incoming subscription items. | ||
| */ | ||
| void subscribe( | ||
| _SubscriptionRequest subscriptionRequest, | ||
| CancelableClientCallStreamObserver<_SubscriptionItem> subscription); | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
111 changes: 111 additions & 0 deletions
111
momento-sdk/src/test/java/momento/sdk/SubscriptionWrapperTest.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,111 @@ | ||
| package momento.sdk; | ||
|
|
||
| import static org.junit.jupiter.api.Assertions.assertFalse; | ||
| import static org.junit.jupiter.api.Assertions.assertTrue; | ||
|
|
||
| import grpc.cache_client.pubsub._Heartbeat; | ||
| import grpc.cache_client.pubsub._SubscriptionItem; | ||
| import grpc.cache_client.pubsub._SubscriptionRequest; | ||
| import io.grpc.Status; | ||
| import io.grpc.StatusRuntimeException; | ||
| import java.util.concurrent.CompletableFuture; | ||
| import java.util.concurrent.Semaphore; | ||
| import java.util.concurrent.atomic.AtomicBoolean; | ||
| import momento.sdk.internal.SubscriptionState; | ||
| import momento.sdk.responses.topic.TopicSubscribeResponse; | ||
| import org.junit.jupiter.api.BeforeEach; | ||
| import org.junit.jupiter.api.Test; | ||
| import org.mockito.MockitoAnnotations; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| public class SubscriptionWrapperTest { | ||
| private final Logger logger = LoggerFactory.getLogger(SubscriptionWrapperTest.class); | ||
|
|
||
| @BeforeEach | ||
| public void setUp() { | ||
| MockitoAnnotations.openMocks(this); | ||
| } | ||
|
|
||
| @Test | ||
| public void testConnectionLostAndRestored() throws InterruptedException { | ||
| SubscriptionState state = new SubscriptionState(); | ||
| TopicSubscribeResponse.Subscription subscription = | ||
| new TopicSubscribeResponse.Subscription(state); | ||
|
|
||
| AtomicBoolean gotConnectionLostCallback = new AtomicBoolean(false); | ||
| AtomicBoolean gotConnectionRestoredCallback = new AtomicBoolean(false); | ||
|
|
||
| Semaphore waitingForSubscriptionAttempt = new Semaphore(0); | ||
|
|
||
| SendSubscribeOptions options = | ||
| new SendSubscribeOptions( | ||
| "cache", | ||
| "topic", | ||
| (message) -> {}, | ||
| () -> {}, | ||
| (err) -> {}, | ||
| () -> { | ||
| logger.info("Got to our connection lost callback!"); | ||
| gotConnectionLostCallback.set(true); | ||
| }, | ||
| () -> { | ||
| logger.info("Got to our connection restored callback!"); | ||
| gotConnectionRestoredCallback.set(true); | ||
| }, | ||
| state, | ||
| subscription); | ||
|
|
||
| IScsTopicConnection connection = | ||
| new IScsTopicConnection() { | ||
| boolean isOpen = true; | ||
| CancelableClientCallStreamObserver<_SubscriptionItem> subscription; | ||
|
|
||
| @Override | ||
| public void close() { | ||
| logger.info("Connection closed"); | ||
| isOpen = false; | ||
| subscription.onError(new StatusRuntimeException(Status.UNAVAILABLE)); | ||
| } | ||
|
|
||
| @Override | ||
| public void open() { | ||
| logger.info("Connection opened"); | ||
| isOpen = true; | ||
| } | ||
|
|
||
| @Override | ||
| public void subscribe( | ||
| _SubscriptionRequest subscriptionRequest, | ||
| CancelableClientCallStreamObserver<_SubscriptionItem> subscription) { | ||
| this.subscription = subscription; | ||
| if (isOpen) { | ||
| _SubscriptionItem heartbeat = | ||
| _SubscriptionItem.newBuilder() | ||
| .setHeartbeat(_Heartbeat.newBuilder().build()) | ||
| .build(); | ||
| subscription.onNext(heartbeat); | ||
| } else { | ||
| subscription.onError(new StatusRuntimeException(Status.UNAVAILABLE)); | ||
| } | ||
| waitingForSubscriptionAttempt.release(); | ||
| } | ||
| }; | ||
|
|
||
| SubscriptionWrapper subscriptionWrapper = new SubscriptionWrapper(connection, options); | ||
| CompletableFuture<Void> subscribeWithRetryResult = subscriptionWrapper.subscribeWithRetry(); | ||
| subscribeWithRetryResult.join(); | ||
|
|
||
| waitingForSubscriptionAttempt.acquire(); | ||
|
|
||
| connection.close(); | ||
|
|
||
| assertTrue(gotConnectionLostCallback.get()); | ||
| assertFalse(gotConnectionRestoredCallback.get()); | ||
|
|
||
| connection.open(); | ||
| waitingForSubscriptionAttempt.acquire(); | ||
|
|
||
| assertTrue(gotConnectionRestoredCallback.get()); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,17 @@ | ||
| <?xml version="1.0" encoding="UTF-8" ?> | ||
| <!DOCTYPE configuration> | ||
|
|
||
| <configuration> | ||
| <import class="ch.qos.logback.classic.encoder.PatternLayoutEncoder"/> | ||
| <import class="ch.qos.logback.core.ConsoleAppender"/> | ||
|
|
||
| <appender name="STDOUT" class="ConsoleAppender"> | ||
| <encoder class="PatternLayoutEncoder"> | ||
| <pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} -%kvp- %msg%n</pattern> | ||
| </encoder> | ||
| </appender> | ||
|
|
||
| <root level="debug"> | ||
| <appender-ref ref="STDOUT"/> | ||
| </root> | ||
| </configuration> |