schulz3000/deepstreamnet

dotnet Client for deepstream.io

deepstreamnet's Issues

Adding an item to a list does not trigger collection changed event

I think my test is ok, but there could be a bug there

        [Test]
        public async Task TestAddingARecordToList()
        {


            var tcs = new TaskCompletionSource<string>();
            var name = Guid.NewGuid().ToString();
            var key = "test_record";
            var listKey = "test_list";
            var c2 = await GetConnection("chris", "test");
            var list2 = await c2.Records.GetListAsync(listKey);
            var keys = list2.ToList();
            // clear work around
            foreach (var recKey in keys)
            {
                list2.Remove(recKey);
            }
            list2.CollectionChanged += (sender, args) =>
            {
                **// never gets here
                if (list2.Any(e => e == key))
                {
                    tcs.SetResult("done");
                }**
            };

            var c = await GetConnection("chris", "test");
            var list = await c.Records.GetListAsync(listKey);
            Assert.IsTrue(list.Count == 0);
            var rec = await c.Records.GetRecordAsync(key);
            rec["name"] = name;
            list.Add(rec.RecordName);
            tcs.Task.Wait();
            c2.Dispose();
            c.Dispose();
        }

Null object exception on listening to list changes

Hi,
with the following test I am getting a null reference on e.OldItems in the JsonNetChangeListener. It occurs on the second call into the CollectionChanged event.

        void Collection_CollectionChanged(object sender, NotifyCollectionChangedEventArgs e)
        {
            if (e.Action == NotifyCollectionChangedAction.Replace || e.Action == NotifyCollectionChangedAction.Remove || e.Action == NotifyCollectionChangedAction.Reset)
            {
                foreach (var item in e.OldItems)

        [Test]
        public async Task TestAddingARecordToList()
        {
            var name = Guid.NewGuid().ToString();
            var key = "test_record";
            var listKey = "test_list";
            using (var watchClient = await GetConnection("chris", "test"))
            {
                var watchList = await watchClient.Records.GetListAsync(listKey);
                var changes = new List<NotifyCollectionChangedEventArgs>();
                watchList.CollectionChanged += (sender, args) =>
                {
                    changes.Add(args);
                };

                using (var updateClient = await GetConnection("admin", "test"))
                {
                    var list = await updateClient.Records.GetListAsync(listKey);

                    var rec = await updateClient.Records.GetRecordAsync(key);
                    var res = await updateClient.Records.SetWithAckAsync(rec, "name", name);
                    Assert.IsTrue(res);
                    list.Add(rec.RecordName);
                }
                while (changes.Count == 0)
                {
                    await Task.Delay(300);
                }

            }
        }

Update WebSocket dependencies

Hello guys,

is it possible to update WebSocket4Net dependency to he latest beta version (0.15.0-beta9) please?

The thing is beside your library I am using another library that has the same dependency on WebSocket4Net but on above mentioned lastest version 0.15.0-beta9. I cannot use them both as the filename is the same and they get overriden. When I try to use your library with the lastest version it gives me an error that it cannot find version 0.15.0.0 (beta6 has .dll version 0.15.0.0 but the lastest beta9 has version 0.15.0.9 and that is the problem).

Many thanks for help.

Trying a basic connection get an Ack timeout

Issue occurs on the following line

       var result = await connection.SendWithAckAsync(Topic.AUTH, Action.REQUEST, Action.Empty, credentials, 1000);

I think my deepstream setup is ok as I can telnet onto the port. I will do some more testing tomorrow with a javascript client to make sure it is not some issue on my deepstream server,

Add / Complete XML Code Comments

Possible bug in DeepStreamRecords

Hey,

I noticed that in this class (code pointer here), you register to three events in the constructor. However on disposing, not only that you don't unregister from these events, you register again.

Is it on purpose? If so, can you explain why do you do it?

DeepStreamRecords - expose RecordUpdated, RecordPatched events

Hi,
I am after an event when a whole record is changed from deepstream. The PropertyChanged is a bit to granular for my need as I only want one event each time you receive updates from deepstream for a record. I am happy to do the pull request.

Record SetWithAckAsync using Path hangs

Seems to hang not getting an ack

        [Test]
        public async Task TestRecordPath()
        {
            var key = "test_record";
            var val = Guid.NewGuid().ToString();
            using (var c = await GetConnection("admin", "test"))
            {
                var rec = await c.Records.GetRecordAsync(key);
                var res = await c.Records.SetWithAckAsync(rec, "details.age", "25");
            }
        }

Property change event on record updated by data provider does not fire

Hi,
if I create data provider and listen for a subscription to a record, then update the record inside the data provider no property change is triggered in the subscribing client. A test is shown below.

    [Test]
    public async Task TestDataProvider()
    {
        var key = @"test/datap/blah";
        var listenKey = @"test/datap/*";
        var val = Guid.NewGuid().ToString();

        // test a data provider

        using (var client = await GetConnection("chris", "test"))
        {
            using (var server = await GetConnection("admin", "test"))
            {
                // create a listener
                await server.Records.ListenAsync(listenKey, async (match, isSubscribed, resposne) =>
                {
                    Assert.AreEqual(key, match, "should be equal");
                    // update the record
                    var rec = await server.Records.GetRecordAsync(match);
                    var obj = new JObject();
                    obj["data"] = val;
                    var result = await server.Records.SetWithAckAsync(rec, obj);
                    if (result)
                        resposne.Accept();
                    else
                        resposne.Reject();
                });

                // get the record, should trigger the data listener
                // to update the object
                var recsub = await client.Records.GetRecordAsync(key);
                int changes = 0;
                // subscribe to record changes
                recsub.PropertyChanged += (sender, args) =>
                {
                    // never gets fired
                    changes++;
                };

                // wait for a change to be triggered
                while (changes == 0)
                {
                    System.Threading.Thread.Sleep(300);

                }
                // never gets here
                Assert.AreEqual(val, recsub["val"].ToString());
            }
        }
    }

List.Clear does not work

A sample unit test that fails

  public async Task TestListClear()
        {
            using (var c = await GetConnection("admin", "test"))
            {
                var key = "test_record";
                var listKey = "test_list";
                var list2 = await c.Records.GetListAsync(listKey);
                list2.Clear();
            }
        }

// from the client
        public void Clear()
        {
            var tmp = innerList;
            innerList.Clear();
            **CollectionChanged?.Invoke(this, new NotifyCollectionChangedEventArgs(NotifyCollectionChangedAction.Reset, null, tmp));**
        }

System.ArgumentException : Constructor supports only the 'Replace' action.
Parameter name: action
at System.Collections.Specialized.NotifyCollectionChangedEventArgs..ctor(NotifyCollectionChangedAction action, IList newItems, IList oldItems)
at DeepStreamNet.DeepStreamList.Clear() in C:\Dev\rtindex\src\dotnet\RtIndex.DeepStream\src\DeepStreamNet\Records\DeepStreamList.cs:line 68
at RtIndex.Tests.DeepStream461.DeepStreamTests.d__9.MoveNext() in C:\Dev\rtindex\src\dotnet\RtIndex.Tests\DeepStream461\DeepStreamTests.cs:line 173
--- End of stack trace from previous location where exception was thrown ---
at NUnit.Framework.Internal.ExceptionHelper.Rethrow(Exception exception)
at NUnit.Framework.Internal.AsyncInvocationRegion.AsyncTaskInvocationRegion.WaitForPendingOperationsToComplete(Object invocationResult)
at NUnit.Framework.Internal.Commands.TestMethodCommand.RunAsyncTestMethod(TestExecutionContext context)

Implement Record "Snapshot" Method

See Javascript Client for Reference

Implement Record "has" Method

See Javascript Client for Reference

Error on websocket: Invalid WebSocket frame: invalid status code 1005

Hello, when I started the ConsoleSample, I could not get the result. The following exception information was found on the server side. How can I deal with this problem?
INFO | Heartbeat missing on websocket, terminating connection
ERROR | Error on websocket: Invalid WebSocket frame: invalid status code 1005

Unknown server protocol version

So with the latest "v2 of this client library, as well as the latest deepstream.io server, I'm getting a unknown server protocol version error.

Seems to happen when I do await _deepStream.LoginAsync(), and throws at await connection.OpenAsync().ConfigureAwait(false);

In Connection.cs, line 57, I did have to comment out client.NoDelay = true;, because the latest prerelease version of WebSocket4Net doesn't have a NoDelay property. Could this have something to do with it?

Problem running .net core 2.1 on linux

Hi,

Just updated on .net core 2.1 and works just great from my dev windows machine when connecting to deepstream 3.10 but when i deploy it to linux (docker) it works randomly. Most of the times stucks in
client.LoginAsync() forever really forever an hour and so.

var client = new DeepStreamClient(host, Convert.ToInt16(port), "deepstream",Convert.ToBoolean(secure));
        
            if (await client.LoginAsync(user, pass))
            {              
                var notifications = await _mediator.Send(new NotificationsQuery.GetNotifications(){UserId = userid});               
                IDeepStreamRecord record = await client.Records.GetRecordAsync($"user/{userid}");               
                await client.Records.SetWithAckAsync(record, notifications);               
            }
            client.Dispose();

Till now was on .net core 2.0 and was working great in production.

Error on Send Event

First of all thank you for having made your deepstream library public.

I had a small problem with the lib regarding sending events, it happens that when the deepstream connection is lost the "Publish" method does not pop an exception and I have not found a way to verify if the connection to WebSocket is active.

To resolve I downloaded your project and made a modification to solve my problem at the moment. Basically I left the connection enumerators exposed so that I can verify that the connection is open and active.

The javascript or deepstream library automatically reconnects whenever the connection is lost.

Sorry for my bad English, because it's difficult to translate things from Portuguese to English with google translator.

How to handle byte arrays with publish/subcribe?

I'm currently trying to see if I can integrate MessagePack (http://msgpack.org) into my app, and while I don't know if deepstream even supports binary data or just strings, I'm successfully packing and unpacking the data.

I can publish the byte data fine, but when I receive it, it comes out as a string in the following format:

2XV7IlZlcnNpb24iOjEsIlRpbWVzdGFtcCI6MTQ4OTUwMDA2NywiVGltZXN0YW1wU3RyaW5nIjoibm93IiwiVHlwZSI6MiwiVHlwZVN0cmluZyI6IkxlYXZlIiwiVVVJRCI6IlBhdWxpIiwiVGV4dCI6bnVsbH0=

The issue is that I don't know how to deserialize this back into a byte array, which I'm hoping you'd be able to help with? :)

Set an array as a new record

Hi,
looks like you can not set an array as a record. The DeepStreamRecordBase must be a JContainer. Should it be a JToken so it can support arrays ? Alternatively throw an exception if you try and set an array.

When you try and set an array this line does not merge anything
Data.Merge(item, jsonMergeSettings);

Here is a test

        [Test]
        public async Task TestSetArray()
        {
            var name = Guid.NewGuid().ToString();
            var key = @"my/testrecord/" + name;
            var obj = new JArray();
            obj.Add("hello");
            using (var client1 = await GetConnection("chris", "test"))
            {
                var record1 = await client1.Records.GetRecordAsync(key);
                var res = await client1.Records.SetWithAckAsync(record1, obj);
                Assert.IsTrue(res);
                var json = JsonConvert.SerializeObject(record1);
                Assert.IsTrue(json.StartsWith("["), "SHould be an array");
            }
        }

record.SetWithAckAsync does not ack

I have a basic test to test save with ack

  public async Task TestAck()
        {
            using (var c = new DeepStreamClient("localhost", 6020))
            {
                var loggedIn = await c.LoginAsync();
                Assert.IsTrue(loggedIn);
                var obj = new JObject();
                obj["name"] = "test me";
                var record = await c.Records.GetRecordAsync("test");
                var saved = await c.Records.SetWithAckAsync(record, obj);
            }
        }

On the setWithAckAsync I get a ack timeout. I have tried increasing the timeout on the client.

the console on the deepstream server looks like
AUTH_ATTEMPT | 127.0.0.1: A�REQ�{}�
AUTH_SUCCESSFUL | open
S | for R:test by open
US | for R:test by open
CLIENT_DISCONNECTED | open

Intermittent error on data provider

Hi, great work on the .net client. Most things seem to be working well. This might not be anything to do with the client but I have this test for creating a data provider. When I run in a sweet of tests it hangs consistently unless I have the 300ms sleep in the test. By itself it seems to be ok. Do you think we need a whenReady method on the record similar to the js client so we can be sure deepstream has done all its things ?

The error on the deepstream console is
INVALID_MESSAGE | R�LA�test/datap/*�test/datap/blah

    [Test]
    public async Task TestDataProvider()
    {
        var val = Guid.NewGuid().ToString();
        var tcs = new TaskCompletionSource<string>();
        var c1 = await GetConnection("admin", "test");
        var c = await GetConnection("admin", "test");
        await c.Records.ListenAsync("test/datap/*", async (match, isSubscribed, resposne) =>
        {

            var rec = await c.Records.GetRecordAsync(match);
            var obj = new JObject();
            obj["data"] = val;
            bool result = await c.Records.SetWithAckAsync(rec, obj);
            await c.Records.DiscardAsync(rec);
            if (result)
            {
                resposne.Accept();
            }
            else
            {
                resposne.Reject();
            }
            tcs.SetResult("done");
        });
        var recsub = await c1.Records.GetRecordAsync("test/datap/blah");
        await tcs.Task;

        // some sort of timing issue�, without the sleep here the test hangs when run in a sweet of tests
        // without the sleep I get INVALID_MESSAGE | R�LA�test/datap/*�test/datap/blah
        // on the deepstream sys out seems to have a funny character which shows up asa box with a question mark
        System.Threading.Thread.Sleep(300);

        // should not have to do this but record is not getting the updates data from the data provider
        // so discard the record and get a snapshot to prove the data was saved
        // may be related to the property change event not firing
        // there is an issue raised on github to fix this
        await c1.Records.DiscardAsync(recsub);
        var snapShot = await c1.Records.SnapshotAsync("test/datap/blah");
        var updatedVal = snapShot["data"];
        Assert.AreEqual(val, updatedVal.ToString());

        c1.Dispose();
        c.Dispose();

    }

Multiple subscriptions hangs client

Kind of a strange situation to do this, I found it accidentally writing a test,

        [Test]
        public async Task TestRecAndSnapshot()
        {
            using (var c = await GetConnection("admin", "test"))
            {
                var snapshot = await c.Records.SnapshotAsync("test/t");
                var rec = await c.Records.GetRecordAsync("test/t");
            }
        }

Subscribe to record property change not triggered on same connection

Here is a quick test that creates a record, sets it with an ack, then subscribes to changes, then sets with an ack again. Finally it waits for the property change event to trigger which never happens.

Note this is all on the one connection.

        [Test]
        public async Task SubscribeToChangesOnSameConnection()
        {
            var key = @"test/helloworld";
            using (var writerClient = await GetConnection("john", "test"))
            {
                var obj = new JObject();
                obj["name"] = "test me";
                var record = await writerClient.Records.GetRecordAsync(key);
                var set = await writerClient.Records.SetWithAckAsync(record, obj);
                Assert.IsTrue(set);
                var events = new List<PropertyChangedEventArgs>();
                record.PropertyChanged += (sender, args) =>
                {
                    events.Add(args);
                };
                obj["name"] = "bob";
                set = await writerClient.Records.SetWithAckAsync(record, obj);
                Assert.IsTrue(set);

               // hangs here and never gets the event
                while (events.Count == 0)
                {
                    System.Threading.Thread.Sleep(200);
                }
            }
        }

Set serialization options

Hi,

First of all congrats for your great implementation.

Is it possible to set JsonConvert options in order to use for example CamelCase serialization in properties?
I am playing with records...

Thanks

Need a c# client for Unity

can we use it with Unity games ?? Unity uses .net 2.0

PropertyChanged not firing on record with different connections

If I have 2 records connected to the same topic on different connections, if I change the record on connection1 should a PropertyChanged get fired on connection 2 ?

Here is my test

        public async Task TestRecordChanges()
        {
            var tcs = new TaskCompletionSource<string>();
            var key = "test_record";
            var c1 = await GetConnection("admin", "test");
            var data = new JArray();
            data.Add("hello");
            var rec = await c1.Records.GetRecordAsync("key");
            rec["data"] = data;
            ((JArray)rec["data"]).Add("test1");
            var c2 = await GetConnection("chris", "test");
            var rec2 = await c2.Records.GetRecordAsync(key);
            rec2.PropertyChanged += (sender, args) =>
            {
                // never gets hit
                tcs.SetResult("done");
            };
            
            ((JArray)rec["data"]).Add("test2");

            while (tcs.Task.Status != TaskStatus.RanToCompletion)
            {
                System.Threading.Thread.Sleep(1000);
            }
            //tcs.Task.Wait();
        }

Events Publish / Subscribe on different connections do not fire

A quick test of pubsub across connections

       [Test]
        public async Task TestPububOnTwoDifferentConnectons()
        {
            var key = "test/pubsub/a";
            var tcs = new TaskCompletionSource<string>();
            var c = await GetConnection("admin", "test");
            var c1 = await GetConnection("admin", "test");

            await c.Events.SubscribeAsync(key,o => 
            {
                // never gets here
                tcs.SetResult("done");
                // assert it is the same object
                var jobj = (JObject)o;
            });

            var obj = new JObject();
            obj["name"] = "bob";
            c1.Events.Publish(key, obj);

            tcs.Task.Wait();
            c1.Dispose();
            c.Dispose();
        }

WebSocket4Net.WebSocket does not contain a definition for NoDelay

DeepStreamNet/Internals/Connection.cs(20,20): Error CS1061: Type `WebSocket4Net.WebSocket' does not contain a definition for `NoDelay' and no extension method `NoDelay' of type `WebSocket4Net.WebSocket' could be found. Are you missing an assembly reference? (CS1061)

dynamic set of Record don't trigger change

rec["root"] = new { p1="valp1", p2 = "valp2" };

this is working and Patch is triggered

dynamic rec = await client.Records.GetRecordAsync("test");
rec.root.p2 = "up2";

this not. Patch is not triggered but should

var rec = await client.Records.GetRecordAsync("test");
rec["root"]["p2"] = "up2";

I think solution is referenced to #17

SocketException: The requested address is not valid in its context.

Hi, so I'm trying to create a deepstream client with Unity, and I installed the await/async asset so it will be 100% similar to the examples given here. but I still can't make it work because of the SocketException..
I tried the example Program.cs as a console app and the connection works fine, so it's not a problem of connectivity or server-related issues.

I realise it's not common to use Unity as a client but it should be fairly possible considering Unity is now .NET 4.x compatible..

So this is my Unity code:

public class DeepstreamManager : MonoBehaviour
{
    private DeepStreamClient client;

    async void Start()
    {
        client = new DeepStreamClient("localhost", 6020);
        if (await client.LoginAsync())
        {
            var eventSubscription = await client.Events.SubscribeAsync("test", x => { print(x); });

            client.Events.Publish("test", "Hello");

            client.Events.Publish("test", 42);
        }
    }
}

And this is the exception:

SocketException: The requested address is not valid in its context.

System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw () (at <9577ac7a62ef43179789031239ba8798>:0)
System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess (System.Threading.Tasks.Task task) (at <9577ac7a62ef43179789031239ba8798>:0)
System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification (System.Threading.Tasks.Task task) (at <9577ac7a62ef43179789031239ba8798>:0)
System.Runtime.CompilerServices.TaskAwaiter.ValidateEnd (System.Threading.Tasks.Task task) (at <9577ac7a62ef43179789031239ba8798>:0)
System.Runtime.CompilerServices.ConfiguredTaskAwaitable+ConfiguredTaskAwaiter.GetResult () (at <9577ac7a62ef43179789031239ba8798>:0)
DeepStreamNet.DeepStreamClient+<LoginAsync>d__17.MoveNext () (at <a91fb4570bc84808b3ad9f1017f460a2>:0)
--- End of stack trace from previous location where exception was thrown ---
System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw () (at <9577ac7a62ef43179789031239ba8798>:0)
System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess (System.Threading.Tasks.Task task) (at <9577ac7a62ef43179789031239ba8798>:0)
System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification (System.Threading.Tasks.Task task) (at <9577ac7a62ef43179789031239ba8798>:0)
System.Runtime.CompilerServices.TaskAwaiter.ValidateEnd (System.Threading.Tasks.Task task) (at <9577ac7a62ef43179789031239ba8798>:0)
System.Runtime.CompilerServices.TaskAwaiter`1[TResult].GetResult () (at <9577ac7a62ef43179789031239ba8798>:0)
DeepstreamManager+<Start>d__1.MoveNext () (at Assets/Scripts/DeepstreamManager.cs:16)
--- End of stack trace from previous location where exception was thrown ---
System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw () (at <9577ac7a62ef43179789031239ba8798>:0)
System.Runtime.CompilerServices.AsyncMethodBuilderCore+<>c.<ThrowAsync>b__6_0 (System.Object state) (at <9577ac7a62ef43179789031239ba8798>:0)
UnityEngine.UnitySynchronizationContext+WorkRequest.Invoke () (at <82ec470c913149358ad0c83ad628910c>:0)
UnityEngine.UnitySynchronizationContext:ExecuteTasks()

Has anyone managed to make Unity work with Deepstream?

Subscribing to record updates/changes

I have noticed that there is no subscribe method for record updates in the DeepStreamRecord class.
I want to know if there is another way to subscribe to a change on a record.

Update to urp protocol

Update client to support urp protocol for deepstream V4

Publish Subscribe on same connection hangs

When I run this test using a different connection for the publish and subscribe it works. On the same connection it hangs.

        [Test]
        public async Task TestPububOnOneConnectons()
        {
            var key = "test/pubsub/a";
            var tcs = new TaskCompletionSource<string>();
            var c = await GetConnection("admin", "test");

            await c.Events.SubscribeAsync(key, o =>
            {
                // never gets here
                tcs.SetResult("done");
                // assert it is the same object
                var jobj = (JObject)o;
            });

            var obj = new JObject();
            obj["name"] = "bob";
            c.Events.Publish(key, obj);

            await tcs.Task;
            c.Dispose();
        }

DeepStreamRecordBase Parse failed when listening to a new list

I got a crash in the Parse method of DeepStreamRecordBase when listening to changes on a new list and adding a item on a different connection. I patched the Parse method to get it working

                var arrayParentPath = path.Substring(0, path.Length - (path.LastIndexOf("[", StringComparison.Ordinal) + 1));
                **if (path.StartsWith("[") && path.EndsWith("]"))
                {
                    arrayParentPath = "";
                }**
                var token = Data.SelectToken(arrayParentPath);
                ((JArray)token).Add(item);

Below is a test to reproduce.

        [Test]
        public async Task TestAddingARecordToListAndListenToChanges()
        {
            var name = Guid.NewGuid().ToString();
            var key = "testrecord/" + Guid.NewGuid().ToString();
            var listKey = "testlist/" + Guid.NewGuid().ToString();
            using (var updateClient = await GetConnection("admin", "test"))
            {
                var list = await updateClient.Records.GetListAsync(listKey);

                var rec = await updateClient.Records.GetRecordAsync(key);
                var res = await updateClient.Records.SetWithAckAsync(rec, "name", name);
                Assert.IsTrue(res);
                using (var readClient = await GetConnection("admin", "test"))
                {

                    var listRead = await readClient.Records.GetListAsync(listKey);
                    var changes = new List<NotifyCollectionChangedEventArgs>();
                    listRead.CollectionChanged += (sender, args) =>
                    {
                        changes.Add(args);
                    };
                    list.Add(rec.RecordName);
                    while (changes.Count == 0)
                    {
                        await Task.Delay(200);
                    }
                    Assert.IsTrue(changes.Count(e => e.Action == NotifyCollectionChangedAction.Add) == 1);
                }
            }
        }

Impliment List Methods

Hi,
does your lib support the deepstream lists ? Could not see it in the api.

Publish Nuget package

Any plans to publish this package on Nuget.org?

Would be greatly appreciated.

How does this constantly listen for messages?

Hi, just looking at this as I've been writing a python client for deepstream.

How does your client constantly listen for messages from the server?

The MessageLoopAsync seems to only be run when you initialise the connection, and then stops reading after no more messages have come through.

You also read from the connection with your SendWithAck, but what happens when the server sends some info about an event later on. How does your client receive it?

Implement Specflow Tests

see Specflow branch for skeleton