jeikabu / nng.netcore Goto Github PK

Hi,

this maybe not a bug, I am new guy with c#.
I have working code in c++ and python, then I connect to c# code use pair protocol.
It can connect but in c# code RecvMsg not return.

If there is a demo code should be nice.
test code has some irrelevant code it's not friendly enough to show how to use.
I use nng.NETCore

c++ pair client code（working）nng-1.3.0
void CnngClientDemoDlg::OnBnClickedBtnStartPair2()
{
nng_socket sock;
char* buf = NULL;
int iRet = 0;
size_t sz;

if ((iRet = nng_pair0_open(&sock)) != 0) {
	ASSERT(FALSE);
}
if ((iRet = nng_dial(sock, URL_PAIR, NULL, 0)) != 0) {
	ASSERT(FALSE);
}
string strSend = "test:123";
nng_msg* msg;
iRet = nng_msg_alloc(&msg, 0);
nng_msg_append(msg, strSend.c_str(), strSend.size());

if ((iRet = nng_sendmsg(sock, msg, 0)) != 0) {
	ASSERT(FALSE);
}
if ((iRet = nng_recv(sock, &buf, &sz, NNG_FLAG_ALLOC)) != 0) {
	ASSERT(FALSE);
}
// This assumes that buf is ASCIIZ (zero terminated).
nng_msg_free(msg);
nng_close(sock);

}

c# pair server (not working)
var path = Path.GetDirectoryName(typeof(nngTest).Assembly.Location);
var ctx = new nng.NngLoadContext(path);
var factory = nng.NngLoadContext.Init(ctx);

        var barrier = new AsyncBarrier(2);

        using (var socket = factory.PairOpen().ThenListenAs(out var listener, pair_url).Unwrap())

or
using (var socket = factory.PairOpen().ThenListen(pair_url).Unwrap())
{
var request = socket.RecvMsg().Unwrap();
var str = Encoding.ASCII.GetString(request.AsSpan());
Console.WriteLine(str);
var reply = factory.CreateMessage();
var replyBytes = Encoding.ASCII.GetBytes("send msg");
reply.Append(replyBytes);
socket.SendMsg(reply);
}

c++ pair server (working with c++ client) nng-1.3.0
int rv;
if ((rv = nng_pair0_open(&pwnd->m_sockPair)) != 0) {
ASSERT(FALSE);
}
if ((rv = nng_listen(pwnd->m_sockPair, URL_PAIR, NULL, 0)) != 0) {
ASSERT(FALSE);
}

while (FLAG_EXIT_PAIR != (pwnd->m_flag & FLAG_EXIT_PAIR))
{
	char* buf = NULL;
	size_t   sz = 0;
	nng_msg* msg = NULL;
	if ((rv = nng_recvmsg(pwnd->m_sockPair, &msg, 0)) != 0)
	{
		if (rv == NNG_ECLOSED)
		{
			break;
		}
		else
		{
			nng_close(pwnd->m_sockPair);
			ASSERT(FALSE);
		}
	}
	string result = (char*)nng_msg_body(msg);
	printf("recv: %s", nng_msg_body(msg));
	// 这里会自动释放 buf
	rv = nng_send(pwnd->m_sockPair, buf, sz, 0);
	if (rv != 0) {
		ASSERT(FALSE);
	}
	nng_msg_free(msg);

}

public IntPtr GetBodyPtr() => nng_msg_body(NngMsg);
or
public T AsStruct() => Marshal.PtrToStructure(nng_msg_body(NngMsg));

NNG has some HTTP client/server functions. Probably doesn't make sense to implement these since .NET Core has ASP.NET Core.

Is there way to get event when the socket connected?
I want to keep the information of client in server, but i can find a way to know which client is connected.

The example just only the client send the data to server and then then server can reply it.

Tests using TCP/WebSocket protocols occasionally fail with EADDRINUSE. Would be nice to use URL like ws://localhost:0 to get a random available port. NNG supports binding to an ephemeral port, but doesn't easily permit discovering it for clients; see nanomsg/nng#830.

https://nanomsg.github.io/nng/man/v1.1.0/nng_pair.7

dotnet build in qemu arm32 (e.g. via scripts/qemu_arm32.sh) fails with:

qemu: uncaught target signal 11 (Segmentation fault) - core dumped

Seems to be known issue with dotnet on qemu arm32.

ReqAsyncCtx/SurveyAsyncContext/etc. are implemented with nng_ctx by inheriting from AsyncCtx<T>. Should change to a "has-a"/composition design to share more implementation with SendReceiveAsyncContext.
Also need tests validating features enabled through nng_ctx use.

Wrapping and tests for nng_stream_* family of functions

Why wasn't it released? nuget

Is this the official nugget package?
https://www.nuget.org/packages/Subor.nng.NETCore/1.1.1

I ask because I cannot find there some classes you have here.

Thanks.

Each protocol has a nng_*_open_raw() function. Need to figure out the best way to implement this. If they have different behavior from the regular sockets should they have their own classes/interfaces?

Both nng_listen and nng_dial accept a NNG_FLAG_NONBLOCK flag. Should support this.

Netcoreapp2.2 target projected after adding reference to Subor.NNG.net doesn't put nng.NETCore.dll in right place. Thus you can't use after restore.

PS Package structure quite unusual, didn't get to the root case.

Created csharp8 branch to enable C#8 and "nullable reference types". Should finalize that for the release of .NET Core 3.

https://nanomsg.github.io/nng/man/v1.1.0/nng_surveyor.7

nanomsg/nng@5803db0

nng_*_getopt_string variants are missing from Listener, Socket, and Dialer.
Strings returned by these functions are allocated via nng_strdup and must be released with nng_strfree.

Pipe has a GetOpt(string name, out string data) but doesn't call nng_strfree.

Follow-up to #71.

Similar to what's done in the Rust wrapper, want to replace TestFactory with multiple implementations of IAPIFactory that create different versions of sockets. Namely, PairOpen() returns a pair0 vs pair1 socket.

I have a simple Req/Rep test using an IReqRepAsyncContext (and IRepReqAsyncContext on the other side).
If I target win-x64 it works like a charm.
If I target x86 it crashes silently (no execption, no output on console) after a few messages have been exchanged.

I do not know if this stems from nng.dll itself or not (problem in nng.NETCore ?problem in nng ? bad version of the library ?).

I use nng.NetCore 1.1.1

Here's the code:

    class Program
    {
        static void Main(string[] args)
        {
            var path = System.IO.Path.GetDirectoryName(typeof(Program).Assembly.Location);
            var ctx = new nng.NngLoadContext(path);
            var factory = nng.NngLoadContext.Init(ctx);

            if (args[0] == "Client")
                Comm.Client(factory).Wait();
            else
                Comm.Server(factory).Wait();
        }
    }

    class Comm
    {
        public static async Task Client(nng.IAPIFactory<nng.IMessage> factory)
        {
            var quit = false;
            using (var socket = factory.RequesterOpen().ThenDial("tcp://localhost:8021").Unwrap())
            using (var Context = factory.CreateReqRepAsyncContext(socket).Unwrap())
            {

                Console.Out.WriteLine("Connected !");
                while (!quit)
                {
                    Console.Write(">");
                    var msg = Console.In.ReadLine();
                    if (!String.IsNullOrEmpty(msg))
                    {
                        quit = msg == "q";
                        using (var request = factory.CreateMessage())
                        {
                            var requestBytes = Encoding.ASCII.GetBytes(msg);
                            request.Append(requestBytes);

                            using (var reply = (await Context.Send(request)).Unwrap())
                            {
                                var str = Encoding.ASCII.GetString(reply.AsSpan().ToArray());
                                Console.WriteLine($"--> '{str}'");
                            }
                        }
                    }
                }
            }
        }

        public static async Task Server(nng.IAPIFactory<nng.IMessage> factory)
        {
            Console.Out.WriteLine("Starting Server...");
            try
            {
                using (var socket = factory.ReplierOpen().ThenListen("tcp://localhost:8021").Unwrap())
                using (var Context = factory.CreateRepReqAsyncContext(socket).Unwrap())
                {
                    var quit = false;
                    var clientCount = 1;
                    while (!quit)
                    {
                        Console.Out.WriteLine("Waiting for message...");
                        using (var request = (await Context.Receive()).Unwrap())
                        {
                            Console.Out.WriteLine("message received...");
                            var str = Encoding.ASCII.GetString(request.AsSpan().ToArray());
                            Console.WriteLine($"'{str}'");
                            if (str == "q")
                            {
                                clientCount -= 1;
                                Console.Out.WriteLine($"{clientCount} clients remaining");
                            }

                            if (clientCount == 0)
                                quit = true;
                        
                            using (var reply = factory.CreateMessage())
                            {
                                var replyBytes = Encoding.ASCII.GetBytes($"Welcome, {str}");
                                reply.Append(replyBytes);
                                var res = (await Context.Reply(reply)).Unwrap();
                            }
                        }
                    }
                }
            }
            catch (Exception ex)
            {
                Console.Error.WriteLine(ex.ToString());
            }

            Console.Out.WriteLine("exiting Server...");
        }
    }

https://nanomsg.github.io/nng/man/v1.1.0/nng_respondent.7

Stressing PairTests::PairShared crashes with:

The active test run was aborted. Reason: Unhandled Exception: System.NullReferenceException: Object reference not set to an instance of an object.

Loop over code similar to:

async Task DoPairShared(string url)
        {
            int numListeners = 2;
            int numDialers = 2;

            var listenerReady = new AsyncBarrier(numListeners + 1);
            var dialerReady = new AsyncBarrier(numListeners + numDialers);
            var cts = new CancellationTokenSource();

            using (var listenSocket = Factory.PairCreate(url, true).Unwrap())
            using (var dialerSocket = Factory.PairCreate(url, false).Unwrap())
            {
                var tasks = new List<Task>();

                // On listening socket create send/receive AIO
                {
                    var task = Task.Run(async () =>
                    {
                        var ctx = listenSocket.CreateAsyncContext(Factory).Unwrap();
                        await listenerReady.SignalAndWait();
                        await dialerReady.SignalAndWait();
                        while (!cts.IsCancellationRequested)
                        {
                            var msg = await ctx.Receive(cts.Token);
                        }
                    });
                    tasks.Add(task);
                    task = Task.Run(async () =>
                    {
                        var ctx = listenSocket.CreateAsyncContext(Factory).Unwrap();
                        await listenerReady.SignalAndWait();
                        await dialerReady.SignalAndWait();
                        while (!cts.IsCancellationRequested)
                        {
                            var _ = await ctx.Send(Factory.CreateMessage());
                        }
                    });
                    tasks.Add(task);
                }

                await listenerReady.SignalAndWait();

                // On dialing socket create send/receive AIO
                {
                    var task = Task.Run(async () =>
                    {
                        var ctx = dialerSocket.CreateAsyncContext(Factory).Unwrap();
                        await dialerReady.SignalAndWait();
                        while (!cts.IsCancellationRequested)
                        {
                            var msg = await ctx.Receive(cts.Token);
                        }
                    });
                    tasks.Add(task);
                    task = Task.Run(async () =>
                    {
                        var ctx = dialerSocket.CreateAsyncContext(Factory).Unwrap();
                        await dialerReady.SignalAndWait();
                        while (!cts.IsCancellationRequested)
                        {
                            var _ = await ctx.Send(Factory.CreateMessage());
                        }
                    });
                    tasks.Add(task);
                }

                await Util.CancelAfterAndWait(tasks, cts, DefaultTimeoutMs);
            }
        }

Hello,

first of all thank you for this project!

I have two projects which use nanomsg for ipc communication (one in Go and the other in Typescript) and now I want to have a third project in C# which uses NNG (this project) to communicate with the other ones.
The Req/Rep pattern works perfectly. I have some difficulties with the Pair pattern:

When I communicate with pair sockets created by NNG everything is fine. Same for pair sockets created by nanomsg.
But if one pair socket is created by nanomsg, and I try to listen or send messages to the pair socket created by NNG (and vice versa), it doesn't work.

I found out that NNG supports a pair version 1 protocol, which nanomsg doesn't (mangos v2) and looking at your code, it seems like the issue could be that you just expose creating pair sockets with version 1 protocol ?

<Factory.cs>
public NngResult<IPairSocket> PairOpen() => Pair1Socket.Open();

I didn't find a way to call Pair0Socket.Open();

It would be awesome to also have the option to create pair sockets which use the version 0 protocol so that the interoperability between NNG and nanomsg is granted.

Thank you for any reply in advance!

nng_pipe

https://github.com/dotnet/BenchmarkDotNet

The compat API exists for backwards compatibility so nanomsg users can migrate to nng. Doesn't make sense to implement this.

Hello,
I tested sending and receiving messages with two .net core console app, it works well.
But when I tested with .net core console app and c++ console app, the connection was established but sending message was blocking and not working and there is no error message.
.net core console code is below.

        var path = Path.GetDirectoryName(typeof(Program).Assembly.Location);
        var ctx2 = new NngLoadContext(path);
        var Factory = NngLoadContext.Init(ctx2);

        string url = "tcp://127.0.0.1:9001";

        var socket = Factory.PairOpen().ThenDial(url).Unwrap();
        var replyBytes = Encoding.ASCII.GetBytes("test");
        var resp=socket.Send(replyBytes);

TLS transport exists as optional feature of NNG. Requires mbedTLS and not exactly sure how much sense this makes since C# has its own TLS implementation.

The various *AsyncContext::callback() functions use TaskCompletionSource::TrySetException() to relay errors (see TrySetNngError() extension method).
Should replace this mechanism with something like Result<TMessage> like we use in runng.

nng_iov used by stream API

nng_parse_url

Take() clears members but needs to protect against calling nng methods with null values

Add support for nng_send/nng_sendmsg and nng_recv/nng_recvmsg

I'm try to using this wrapper. I follow the examples but got this error.

'NngResult<IRepSocket>' does not contain a definition for 'listenOrDial' and no accessible extension method 'listenOrDial' accepting a first argument of type 'NngResult<IRepSocket>' could be found.

Is it i missing something? Or there has a new examples project?

I am trying to achieve full duplex (concurrent send and receive) operations on a context.

My goal is to be able to serve multiple clients (hence the use of contexts) allowing concurrent send and receive operations.

However, if I try to

• Create a Pair Socket
• Listen
• Create several SendReceiveAcyncContext
• Perform both Send and Receive operations at the same time
  I get an InvalidOperationException. The code indicate that this Context indeed disallows performing both send and receive operations. Is that a hard requirement of nng itself ?

I can achieve what I want using one pair socket, but if I want to accept many clients I need to instantiate as many contexts as needed, if I am not mistaken.

Is there a way to achieve what I want to do ? Is this an intrisic limitation of the Pair socket or Context ?

Some code example:
server:

        ...
        using (var socket = factory.PairOpen().Unwrap())
        {
            var listener = socket.ListenWithListener("tcp://localhost:6789").Unwrap();

            listener.Start();

            for(var i=0; i<MAX_CLIENT_COUNT;i++) {
                var context = factory.CreateSendReceiveAsyncContext(socket, SendReceiveContextSubtype.Pair).Unwrap();
                AddServer(context); //Server will perform concurrent context.Receive() & Context.Send() operations
            }
         }
        ...

client:

     using (var socket = factory.PairOpen().Unwrap())
      {
            Console.WriteLine("Connecting to server...");
            socket.Dial("tcp://localhost:6789");
            Console.WriteLine("Connected to server...");
            var context = factory.CreateSendReceiveAsyncContext(socket, SendReceiveContextSubtype.Pair).Unwrap();
            
            using(Client = new Client(context)) // Client will perform concurrent context.Receive() and context.Send() operations
            {
                        ...
            }
      }

ZeroTier transport is optional NNG feature. Not sure if anyone needs this.

Supplemental functions related to threading and OS functionality. C# has own functions.

I encountered an issue where I get "malloc(): memory corruption (fast)" or other similar errors (using .NET Core 3.1 on Linux) when using ISendSocket.SendMsg(IMessage). Here's the minimal program that will reproduce the issue:

using System.Threading.Tasks;

using nng;
using System.IO;

namespace NngBug
{
    class Program
    {
        static void Main(string[] args)
        {
            var path = Path.GetDirectoryName(typeof(Program).Assembly.Location);
            var ctx = new nng.NngLoadContext(path);
            var factory = nng.NngLoadContext.Init(ctx);
            var addr = "tcp://127.0.0.1:5000";
            using (var repSocket = factory.ReplierOpen().ThenListen(addr).Unwrap())
            using (var reqSocket = factory.RequesterOpen().ThenDial(addr).Unwrap())
            {
                Task.Run(() =>
                {
                    while (true)
                    {
                        // the code block causing the issue
                        using (var msg = factory.CreateMessage())
                        {
                            msg.Append(new byte[100]);
                            reqSocket.SendMsg(msg).Unwrap();
                        }
                        // if I use this instead it works without errors:
                        // reqSocket.Send(new byte[100]).Unwrap();
                        using (var resp = reqSocket.RecvMsg().Unwrap())
                        {
                        }
                    }
                });
                while (true)
                {
                    using (var msg = repSocket.RecvMsg().Unwrap())
                    {
                        repSocket.Send(new byte[100]);
                    }
                }
            }
        }
    }
}

Hi guys,

I am trying to repeat your Publisher/Subscriber example, but with the test examples is not enough to get how to properly do it. Then, could you provide any example (no test), please?

Thanks.

Travis build on linux gets stuck on a test frequently.

Tried adding:

[assembly: CollectionBehavior(DisableTestParallelization = true, MaxParallelThreads = 1)]

Usually after all PushPullTests have passed. If just run ReqRepTests, sometimes ReqRepBasic(ipc/tcp) or nothing complete. Doesn't matter if I comment out ReqRepBasic or ReqRepTasks it still hangs. Probably a general problem with request/reply.

*AsyncContext.Create() methods take ISocket when they should accept more specific ISendSocket, etc.
Also, replace nng_send_aio and nng_recv_aio with calls with wrapped versions like was done for INngCtx.Send()/nng_ctx_send

With a simple client/server test program I encounter a problem:

Server:

using nng;
using System;
using System.IO;
using System.Text;

namespace NNGServer
{
    class Program
    {
        static void Main(string[] args)
        {
            var path = Path.GetDirectoryName(typeof(Program).Assembly.Location);
            var ctx = new nng.NngLoadContext(path);
            var factory = nng.NngLoadContext.Init(ctx);

            var quit = false;
            using (var socket = factory.RequesterOpen().ThenDial("ipc://test.ipc").Unwrap())
            using(var Context = factory.CreateReqRepAsyncContext(socket).Unwrap())
            {
                while(!quit)
                {
                    Console.Write(">");
                    var msg = Console.In.ReadLine();
                    if (!String.IsNullOrEmpty(msg))
                    {
                        quit = msg == "q";
                        using (var request = factory.CreateMessage())
                        {
                            var requestBytes = Encoding.ASCII.GetBytes(msg);
                            request.Append(requestBytes);
                            using (var reply = Context.Send(request).Result.Unwrap())
                            {
                                var str = Encoding.ASCII.GetString(reply.AsSpan());
                                Console.WriteLine($"--> '{str}'");
                            }
                        }
                    }
                }
            }
        }
    }
}

Client:

using nng;
using System;
using System.IO;
using System.Text;

namespace NNGServer
{
    class Program
    {
        static void Main(string[] args)
        {
            var path = Path.GetDirectoryName(typeof(Program).Assembly.Location);
            var ctx = new nng.NngLoadContext(path);
            var factory = nng.NngLoadContext.Init(ctx);

            var quit = false;

            using (var socket = factory.ReplierOpen().ThenListen("ipc://test.ipc").Unwrap())
            using(var Context = factory.CreateRepReqAsyncContext(socket).Unwrap())
            {
                while(!quit)
                {
                    using (var request = Context.Receive().Result.Unwrap())
                    {
                        var str = Encoding.ASCII.GetString(request.AsSpan());
                        Console.WriteLine($"'{str}'");
                        quit = str == "q";
                        using (var reply = factory.CreateMessage())
                        {
                            var replyBytes = Encoding.ASCII.GetBytes($"Welcome, {str}");
                            reply.Append(replyBytes);
                            Context.Reply(reply);
                        }
                    }
                }
            }
        }
    }    
}

Running on .net Core 3.0 under Windows 10 x64, compiled as Debug/AnyCPU.
Using latest Nuget Package (1.1.1).

After the second message sent by the client,

1. The server receives an additionnal, empty message
2. the server abruptly stops with the following message in the Console

panic: appending node already on a list or not inited
This message is indicative of a BUG.
Report this at https://github.com/nanomsg/nng/issues

When running the program using Visual Studio 2019 Debugger, I get the following message on top:

\NNGServer.exe (process 29732) exited with code -1073740791.

Note that under the debugger, no exception is raised. The server program just stops upon reaching the end of the using block for the request IMessage instance.

Moreover, not disposing any IMessage instance leads to

1. Server receiving the first message twice.
2. Server crashing with the same message.

I'm not sure if it was changed recently in VS, it seems to me that i have used previous versions without such problems. But with VS 19.2 it tries to convert projects for no reason and ends up with broken project files.

It seems including common.targets doesn't work as expected. I have fixed projects manually, but the long term solution seems to switch to Directory.Build.props and Directory.Build.targets.

Some contextx support options (for example ReqRep support NNG_OPT_REQ_RESENDTIME ).
There is currently no way to set this option.

1. Contexts should at least support IGetOption/ISetOption
2. Protocol-specific options may be available on the corresponding context with properly named methods (as per SetTimeout on AsyncContext). For example void IAsyncReqRepContext.SetResendTime(int duration)

Our team recently run into a dispose bug, which is fixed by now:
nanomsg/nng@c726f93
For this bug alone It would be good to update the wrapper.

Hi,

I just downloaded the project, opened the solution and when building the Nuget is not able to get any library. If I try to "restore Nuget packages" I just get "Error occurred while restoring NuGet packages: '$(NETStandardImplicitPackageVersion)' is not a valid version string."

I tried switching from subor.NNG.NETCore to nng.NET (1.3.1-rc0) and my .net Core programs fail to load with an exception stating it cannot load the nng.NETCore.dll assembly.

It seems the code for AssemblyLoadContext still mentions nng.NETCOre.dll, when the new dll bears the nng.NET.dll name.

Is it a bug or did something go wrong in my packages when I switched ?

ISockets.cs has a interface IHasOpts with various GetOpt() and SetOpt() methods. Currently it's used by ISocket, IListener, and IDialer. It needs to get split into separate IGetOpt and ISetOpt interfaces because some nng objects (like pipes) have no "setopt" functions.

ISetOpt interface can then be added to IPipe.

Could you give an example on how to use the library properly in .Net 4.7.2 or .Net 4.8 ?
I managed to create an test program that works in .Net Core but the same code crashes silently (app dies without exception or message) when running under .Net 4.7.2 or 4.8.

I wonder if it is linked to properly loading the libraries and setting up the native dependencies.

Any clue ?