feat: add topics #468

nand4011 · 2023-09-01T05:35:44Z

Add the initial topics implementation. The entry point is TopicClient, which contains the publish and subscribe methods. Subscribe returns an IAsyncEnumerable that can be iterated over to read from the topic.

src/Momento.Sdk/Internal/ScsTopicClient.cs

tests/Integration/Momento.Sdk.Tests/TopicTest.cs

Add the initial topics implementation. The entry point is TopicClient, which contains the publish and subscribe methods. Subscribe returns an IAsyncEnumerable<TopicMessage> that can be iterated over to read from the topic.

tests/Integration/Momento.Sdk.Tests/TopicTest.cs

cprice404 · 2023-09-01T15:29:13Z

tests/Integration/Momento.Sdk.Tests/TopicTest.cs

+        var subscribeResponse = await topicClient.SubscribeAsync(cacheName, topicName);
+        Assert.True(subscribeResponse is TopicSubscribeResponse.Subscription,
+            $"Unexpected response: {subscribeResponse}");
+        var subscription = (TopicSubscribeResponse.Subscription)subscribeResponse;


Is this what we are expecting the end user code to look like? They will have to cast it?

I know that with the cache methods we were doing like instanceof Subscription newVarName and then it would smart cast. If there's not going to be a similar way for users to consume this then we may want to tweak how we are exposing the API, I don't want users to have to cast in the happy path.

You can cast in an if statement:
if (response is TopicPublishResponse.Success successResponse) { // handle success as appropriate }
That's what I would expect end user code to look like. I can make the tests look more like that if desired.

cprice404 · 2023-09-01T15:31:56Z

tests/Integration/Momento.Sdk.Tests/TopicTest.cs

+                Assert.True(message is TopicMessage.Item, $"Unexpected message: {message}");
+                if (((TopicMessage.Item)message).ValueString!.Equals(skipMessage)) continue;
+
+                Assert.Equal(valuesToSend[messageCount], ((TopicMessage.Item)message).ValueString);


same here.

i don't mind casting in tests but I don't want it to be the UX.

If this pattern is just for tests then it might be worth adding an example main that shows what you are expecting it to look for users that are consuming it.

The number 1 thing that we have to get correct before we merge this PR is the user-facing API. Everything else can be tweaked/improved/cleaned up later - so you should make sure there is something in the PR that clearly illustrates what you are expecting that to look like for a consumer.

kvcache · 2023-09-01T16:50:50Z

is this missing an assembly reference?

+++ b/tests/Integration/Momento.Sdk.Tests/Momento.Sdk.Tests.Integration.csproj
+  <ItemGroup>
+    <Reference Include="Microsoft.Bcl.AsyncInterfaces, Version=7.0.0.0, Culture=neutral, PublicKeyToken=cc7b13ffcd2ddd51" />
+  </ItemGroup>

kvcache

Took a short look and poked at it in visual studio. Nice docs, and I like how we got to await foreach (var message in subscription)

kvcache · 2023-09-01T17:10:01Z

src/Momento.Sdk/Responses/TopicMessage.cs

+        public Item(_TopicItem topicItem)
+        {
+            _value = topicItem.Value;
+            TopicSequenceNumber = topicItem.TopicSequenceNumber;
+        }


so, oneofs in protocol buffers are a little different for each language, but they always expect some kind of matching before accessing a property. As-is, a user can't tell the difference between a published empty string and a binary message.
We should provide a way for users to be able to tell the difference, possibly similar to the way protocol buffers does it (that's where we got the go matching pattern), between an Error, a Discontinuity, a Text value and a Binary value. Here is the documentation for inspiration: https://learn.microsoft.com/en-us/dotnet/architecture/grpc-for-wcf-developers/protobuf-any-oneof#oneof

public Item(_TopicItem topicItem) { switch (topicItem.Value.KindCase) { case _TopicValue.KindOneofCase.Text: { // The only populated field in this message's value is Text. // Let's vend it to the user that way instead of leaving them to guess? break; } case _TopicValue.KindOneofCase.Binary: { // The only populated field in this message's value is Binary. break; } case _TopicValue.KindOneofCase.None: { // Skip message? It's not useful, and the server did something bad break; } default: { // Skip message? It's not useful, as it holds something undefined break; } } _value = topicItem.Value; TopicSequenceNumber = topicItem.TopicSequenceNumber; }

I agree with that. I was trying to follow the pattern set by the cache methods, where the value is derived from the same ByteString regardless of whether it is text or binary. Since the difference is more explicit here I'm in favor of making them separate types.

@kvcache so do you really think that's the right API here? To force users to always have a switch statement in order to read anything at all off of the subscription, even for the simplest use cases?

that feels overbearing and in the other SDKs I think we have been trying to avoid forcing that on them.

@kvcache random thought:

In the JS SDK we implement subscriptions by allowing the user to pass callbacks for OnItem and OnError. Presumably in the future there might also be OnDiscontinuity.

We did that in JS SDK because that's how the grpc streaming API was exposed.

In golang we did it iterator-style because that's how the grpc streaming API was exposed.

But would it be crazy, in .NET, to do it the callback way?

We could encapsulate the foreach loop inside the SDK in an async function, and allow the user to pass us callbacks for OnBinary, OnText, OnDiscontinuity.

Then they wouldn't have to have a switch statement.

It seems like it might be a better API IMO, with two questions in my mind:

Is it going to be obvious to them that the overloaded signatures of publish will map to the specific OnBinary/OnText callbacks? (This issue is relevant now, even if we stick with the iterator-style API.)

If they need to do anything that has to do with ordering between OnText and OnDiscontinuity, having those separated into two user-provided lambdas might make it harder for them to manage/share state across the two callbacks? maybe not though, they could close over it or use a class that had the two callback functions?

(edit: crossposted)
@cprice404 this is inside the Item class - I do think it is reasonable to expect Momento developers to switch to do the needful with each message on the subscription :-).

Separately, I do not think all users should be forced to use pattern matching if they are of the exceptional persuasion. I tried to suggest that we should have accessors available that throw if the message is not of the required disposition. I think the item type should BOTH [1] have a switch pattern available AND [2] have exceptional typed accessors available.

oh dude, you sparked a memory: I think we might want to think about using delegates for this. You hit on something interesting here.

delegate void TextHandler(string message); delegate void BinaryHandler(byte[] message); class SubscriptionActions { public SubscriptionActions(TextHandler? onText = null, BinaryHandler? onBinary = null) { OnTextMessage += Log; OnBinaryMessage += Log; if (onText != null) { OnTextMessage += onText; } if (onBinary != null) { OnBinaryMessage += onBinary; } } public event TextHandler OnTextMessage; public event BinaryHandler OnBinaryMessage; private void Log<T>(T message) { } } class Lol { SubscriptionActions MakeSubscriptionActionsThatJustPrintTextMessagesToTheConsole() { return new SubscriptionActions(onText: Console.WriteLine); } }

Hmm, as nice as this could be, I think await foreach is too ergonomic to ignore.

src/Momento.Sdk/Responses/TopicMessage.cs

kvcache · 2023-09-01T17:27:07Z

tests/Integration/Momento.Sdk.Tests/TopicTest.cs

+        var subscribeResponse = await topicClient.SubscribeAsync(cacheName, topicName);
+        Assert.True(subscribeResponse is TopicSubscribeResponse.Subscription,
+            $"Unexpected response: {subscribeResponse}");
+        var subscription = (TopicSubscribeResponse.Subscription)subscribeResponse;


Yeah, casting being the pattern is not great. I think you should consider:

Make TopicSubscribeResponse the return type.

Make TopicSubscribeResponse a simple class.

Un-nest response types.

Compose 2 things in the TopicSubscribeResponse:

An enum TopicSubscribeResponse.Kind

A field for each response kind (like protocol buffers, sound familiar?)

Add a couple properties to TopicSubscribeResponse:

Kind - documented so people can choose to switch on it if they want to avoid the throws below...

Subscription - gives you the Subscription or throws a good error about the error.

Error - gives you the Error or throws a good error about how this was not an error.

Casting is not the normal pattern outside of testing. In use it would look more like:

if (subscribeResponse is TopicSubscribeResponse.Subscription subscription) { await foreach (var message in subscription) { switch (message) { case TopicMessage.Binary binary: break; case TopicMessage.Text text: break; case TopicMessage.Discontinuity discontinuity: break; case TopicMessage.Error: break; } } } else if (subscribeResponse is TopicSubscribeResponse.Error error) { // error handling }

That's assuming we split text and binary messages and add discontinuity.

@cprice404 Is there a good place to add an example like this without releasing and adding it to the examples directory? I can make something separate if we want realistic looking code before the examples are ready.

Oh yeah, that's pretty cool. It looks like a type switch + lexical cast. I couldn't tell that's how it was going :-). Flattening the value cases seems reasonable to me too.

re: how to show examples, generally I will just add a throwaway file with the example code, and remove it before the PR is merged. Or add a temporary dependency on ../blah in the Examples project and include an example there, and stash it before the final PR merge.

or pasting an example into github like you did is fine too.

The rewritten tests have a consume function that looks much more like what standard use is:

client-sdk-dotnet/tests/Integration/Momento.Sdk.Tests/TopicTest.cs

Line 171 in 9ad31e5

private async Task<List<TopicMessage>> ConsumeMessages(string topicName, CancellationToken token)

kvcache · 2023-09-01T17:30:54Z

src/Momento.Sdk/Internal/ScsTopicClient.cs

+                    case _SubscriptionItem.KindOneofCase.Discontinuity:
+                        _logger.LogTraceTopicMessageReceived("discontinuity", cacheName, topicName);
+                        break;


this should probably be returning a subscription item so users can be aware of missing messages.

I was going to add Discontinuity as a separate item in a second pass, but I can add it now if if won't add too much complexity to the pr.

yes we need to get this out the door, we will add those later

Split Item into separate Text and Binary types so there is no ambiguity about the type of message received. Add AsyncInterfaces to the integration test csproj.

Add cancellations to the topic tests so guard against deadlocks. Temporarily change the ubuntu test runner to macos to see if the hanging is os dependent.

nand4011 · 2023-09-01T20:31:23Z

ubuntu-latest and windows-latest seem to hang forever when running the integration tests, while macos-latest passes. The tests run locally on mac os and amazon linux. Continuing to troubleshoot.

nand4011 · 2023-09-01T20:55:36Z

This could be a problem:
C:\Users\runneradmin\.nuget\packages\microsoft.bcl.asyncinterfaces\7.0.0\buildTransitive\net461\Microsoft.Bcl.AsyncInterfaces.targets(4,5): warning : Microsoft.Bcl.AsyncInterfaces 7.0.0 doesn't support net461 and has not been tested with it. Consider upgrading your TargetFramework to net462 or later. You may also set <SuppressTfmSupportBuildWarnings>true</SuppressTfmSupportBuildWarnings> in the project file to ignore this warning and attempt to run in this unsupported configuration at your own risk. [D:\a\client-sdk-dotnet\client-sdk-dotnet\src\Momento.Sdk\Momento.Sdk.csproj]

Add a net6.0 windows target to the github tests.

cprice404 · 2023-09-02T00:32:18Z

src/Momento.Sdk/TopicClient.cs

+    /// </summary>
+    /// <param name="config">Configuration to use for the transport, retries, middlewares. See <see cref="Configurations"/> for out-of-the-box configuration choices, eg <see cref="Configurations.Laptop.Latest"/></param>
+    /// <param name="authProvider">Momento auth provider.</param>
+    public TopicClient(IConfiguration config, ICredentialProvider authProvider)


We do not want to use the CacheClient Configuration object here. We need a new TopicConfiguration interface/object with the minimal surface area.

Remove top level value methods and the sequence number from TopicMessage. Add pre-built topic configurations. Remove max requests and eager connections from the topic manager. Separate enumerator and subscription cancellation.

nand4011 requested review from cprice404, malandis and a team September 1, 2023 05:36

nand4011 force-pushed the initial-topics branch from 65d1fe4 to 62e3094 Compare September 1, 2023 05:40

nand4011 commented Sep 1, 2023

View reviewed changes

src/Momento.Sdk/Internal/ScsTopicClient.cs Outdated Show resolved Hide resolved

nand4011 commented Sep 1, 2023

View reviewed changes

tests/Integration/Momento.Sdk.Tests/TopicTest.cs Outdated Show resolved Hide resolved

feat: add topics

e10ae02

Add the initial topics implementation. The entry point is TopicClient, which contains the publish and subscribe methods. Subscribe returns an IAsyncEnumerable<TopicMessage> that can be iterated over to read from the topic.

nand4011 force-pushed the initial-topics branch from 62e3094 to e10ae02 Compare September 1, 2023 06:37

cprice404 requested a review from kvcache September 1, 2023 15:19

cprice404 reviewed Sep 1, 2023

View reviewed changes

kvcache reviewed Sep 1, 2023

View reviewed changes

nand4011 added 2 commits September 1, 2023 11:31

split message types

4f4cb06

Split Item into separate Text and Binary types so there is no ambiguity about the type of message received. Add AsyncInterfaces to the integration test csproj.

Return false from the iterator when the stream is cancelled.

5e1f623

Add cancellations to the topic tests so guard against deadlocks. Temporarily change the ubuntu test runner to macos to see if the hanging is os dependent.

Add print statements to try and identify hanging test.

f4623d1

nand4011 added 4 commits September 1, 2023 15:17

Only enable topics for .NET Standard 2.0 or greater

b728deb

Add a net6.0 windows target to the github tests.

Comment out tests for troubleshooting and replace sleep with delay

117257e

Manually synchronize the test and add more print statements

fc66489

Temporarily remove subscribing from test.

be310b9

cprice404 suggested changes Sep 2, 2023

View reviewed changes

nand4011 added 8 commits September 1, 2023 17:38

Make the test as simple as possible.

541b0e2

Modify the test to only create a task that waits.

68b1cf3

Modify the test to publish messages.

34180b8

Remake the test to be explicitly threaded

8382d43

Remake publish/subscribe tests to be explicitly threaded.

9ad31e5

Make message production thread in tests.

264d46f

re-add the single test filter

d7f8738

Troubleshoot test threading.

eedc1ad

nand4011 and others added 10 commits September 5, 2023 10:28

Remove test filter from build.yaml

f710074

add and use a separate topic configuration.

8252472

detailed logging

020c364

switch to normal verbosity, & cancel even if something goes wrong

a8044e5

xunit uses a different way to set up verbosity

f0e603c

naming, threading, test case coverage

b56bfb4

Temporarily disable one of the topic tests.

9182598

leave continuation associated with original async context

0180c5b

disable test parallelism

9e49d9f

Add ValueString and ValueByteArray methods to TopicMessage.

0b81e15

nand4011 requested review from cprice404 and kvcache September 5, 2023 23:13

Multiple pr fixes

052d9cd

Remove top level value methods and the sequence number from TopicMessage. Add pre-built topic configurations. Remove max requests and eager connections from the topic manager. Separate enumerator and subscription cancellation.

cprice404 approved these changes Sep 6, 2023

View reviewed changes

nand4011 merged commit 5ce2ff7 into main Sep 6, 2023
7 checks passed

nand4011 deleted the initial-topics branch September 6, 2023 19:43

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: add topics #468

feat: add topics #468

nand4011 commented Sep 1, 2023

cprice404 Sep 1, 2023

nand4011 Sep 1, 2023 •

edited

Loading

cprice404 Sep 1, 2023

kvcache commented Sep 1, 2023

kvcache left a comment

kvcache Sep 1, 2023

nand4011 Sep 1, 2023

cprice404 Sep 2, 2023

cprice404 Sep 2, 2023

kvcache Sep 2, 2023 •

edited

Loading

kvcache Sep 2, 2023

kvcache Sep 2, 2023 •

edited

Loading

kvcache Sep 2, 2023

kvcache Sep 1, 2023

nand4011 Sep 1, 2023

nand4011 Sep 1, 2023

kvcache Sep 1, 2023

cprice404 Sep 2, 2023

nand4011 Sep 2, 2023

kvcache Sep 1, 2023

nand4011 Sep 1, 2023

cprice404 Sep 2, 2023

nand4011 commented Sep 1, 2023

nand4011 commented Sep 1, 2023

cprice404 Sep 2, 2023

feat: add topics #468

feat: add topics #468

Conversation

nand4011 commented Sep 1, 2023

Choose a reason for hiding this comment

nand4011 Sep 1, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kvcache commented Sep 1, 2023

kvcache left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kvcache Sep 2, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kvcache Sep 2, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

nand4011 commented Sep 1, 2023

nand4011 commented Sep 1, 2023

Choose a reason for hiding this comment

nand4011 Sep 1, 2023 •

edited

Loading

kvcache Sep 2, 2023 •

edited

Loading

kvcache Sep 2, 2023 •

edited

Loading