Merge pull request #1 from nats-io/add_cs_client

Initial NATS C# Client (Alpha)

LICENSE

@@ -0,0 +1,20 @@
+The MIT License (MIT)
+
+Copyright (c) 2012-2015 Apcera Inc.
+
+Permission is hereby granted, free of charge, to any person obtaining a copy of
+this software and associated documentation files (the "Software"), to deal in
+the Software without restriction, including without limitation the rights to
+use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
+the Software, and to permit persons to whom the Software is furnished to do so,
+subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
+IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

NATS.sln

@@ -0,0 +1,56 @@
+
+Microsoft Visual Studio Solution File, Format Version 12.00
+# Visual Studio Express 2012 for Windows Desktop
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "NATS", "NATS\NATS.csproj", "{68EE71D4-9532-470E-B5CA-ECAA79936B1F}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "NATSUnitTests", "NATSUnitTests\NATSUnitTests.csproj", "{00DBFD4D-72F4-4250-884C-C1527C66A0C2}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Publish", "examples\Publish\Publish.csproj", "{A434BE66-EC4D-4FA4-89C7-9097A22319FF}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Subscribe", "examples\Subscribe\Subscribe.csproj", "{0D44FEE5-87D7-4DC5-956C-B03C3A5B286F}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "QueueGroup", "examples\QueueGroup\QueueGroup.csproj", "{E8AECF76-0A83-4128-B2AA-27CBDA9CCA50}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Requestor", "examples\Requestor\Requestor.csproj", "{6C2C965C-D1F9-4E5E-863C-EE5C51937866}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Replier", "examples\Replier\Replier.csproj", "{BEC6E07B-3DFC-493F-82E6-35A32D52ED2D}"
+EndProject
+Global
+	GlobalSection(SolutionConfigurationPlatforms) = preSolution
+		Debug|Any CPU = Debug|Any CPU
+		Release|Any CPU = Release|Any CPU
+	EndGlobalSection
+	GlobalSection(ProjectConfigurationPlatforms) = postSolution
+		{68EE71D4-9532-470E-B5CA-ECAA79936B1F}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{68EE71D4-9532-470E-B5CA-ECAA79936B1F}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{68EE71D4-9532-470E-B5CA-ECAA79936B1F}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{68EE71D4-9532-470E-B5CA-ECAA79936B1F}.Release|Any CPU.Build.0 = Release|Any CPU
+		{00DBFD4D-72F4-4250-884C-C1527C66A0C2}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{00DBFD4D-72F4-4250-884C-C1527C66A0C2}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{00DBFD4D-72F4-4250-884C-C1527C66A0C2}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{00DBFD4D-72F4-4250-884C-C1527C66A0C2}.Release|Any CPU.Build.0 = Release|Any CPU
+		{A434BE66-EC4D-4FA4-89C7-9097A22319FF}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{A434BE66-EC4D-4FA4-89C7-9097A22319FF}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{A434BE66-EC4D-4FA4-89C7-9097A22319FF}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{A434BE66-EC4D-4FA4-89C7-9097A22319FF}.Release|Any CPU.Build.0 = Release|Any CPU
+		{0D44FEE5-87D7-4DC5-956C-B03C3A5B286F}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{0D44FEE5-87D7-4DC5-956C-B03C3A5B286F}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{0D44FEE5-87D7-4DC5-956C-B03C3A5B286F}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{0D44FEE5-87D7-4DC5-956C-B03C3A5B286F}.Release|Any CPU.Build.0 = Release|Any CPU
+		{E8AECF76-0A83-4128-B2AA-27CBDA9CCA50}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{E8AECF76-0A83-4128-B2AA-27CBDA9CCA50}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{E8AECF76-0A83-4128-B2AA-27CBDA9CCA50}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{E8AECF76-0A83-4128-B2AA-27CBDA9CCA50}.Release|Any CPU.Build.0 = Release|Any CPU
+		{6C2C965C-D1F9-4E5E-863C-EE5C51937866}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{6C2C965C-D1F9-4E5E-863C-EE5C51937866}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{6C2C965C-D1F9-4E5E-863C-EE5C51937866}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{6C2C965C-D1F9-4E5E-863C-EE5C51937866}.Release|Any CPU.Build.0 = Release|Any CPU
+		{BEC6E07B-3DFC-493F-82E6-35A32D52ED2D}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{BEC6E07B-3DFC-493F-82E6-35A32D52ED2D}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{BEC6E07B-3DFC-493F-82E6-35A32D52ED2D}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{BEC6E07B-3DFC-493F-82E6-35A32D52ED2D}.Release|Any CPU.Build.0 = Release|Any CPU
+	EndGlobalSection
+	GlobalSection(SolutionProperties) = preSolution
+		HideSolutionNode = FALSE
+	EndGlobalSection
+EndGlobal

NATS/AsyncSub.cs

@@ -0,0 +1,142 @@
+// Copyright 2015 Apcera Inc. All rights reserved.
+
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+// disable XML comment warnings
+#pragma warning disable 1591
+
+namespace NATS.Client
+{
+    /// <summary>
+    /// An object of this class is an asynchronous subscription representing interest
+    /// in a subject.   The subject can have wildcards (partial:*, full:>).
+    /// Messages will be delivered to the associated MsgHandler event delegates.
+    /// While nothing prevents event handlers from being added or 
+    /// removed while processing messages, no messages will be received until
+    /// Start() has been called.  This allows all event handlers to be added
+    /// before message processing begins.
+    /// </summary>
+    /// <remarks><see cref="MsgHandler">See MsgHandler</see>.</remarks>
+    public sealed class AsyncSubscription : Subscription, IAsyncSubscription, ISubscription
+    {
+
+        public MsgHandler          msgHandler = null;
+        private MsgHandlerEventArgs msgHandlerArgs = new MsgHandlerEventArgs();
+        private Task                msgFeeder = null;
+
+        internal AsyncSubscription(Connection conn, string subject, string queue)
+            : base(conn, subject, queue) { }
+
+        internal protected override bool processMsg(Msg msg)
+        {
+            Connection c;
+            MsgHandler handler;
+            long max;
+
+            lock (mu)
+            {
+                c = this.conn;
+                handler = this.msgHandler;
+                max = this.max;
+            }
+
+            // the message handler has not been setup yet, drop the 
+            // message.
+            if (msgHandler == null)
+                return true;
+
+            if (conn == null)
+                return false;
+
+            long d = Interlocked.Increment(ref delivered);
+            if (max <= 0 || d <= max)
+            {
+                msgHandlerArgs.msg = msg;
+                try
+                {
+                    msgHandler(this, msgHandlerArgs);
+                }
+                catch (Exception) { }
+
+                if (d == max)
+                {
+                    Unsubscribe();
+                    this.conn = null;
+                }
+            }
+
+            return true;
+        }
+
+        internal bool isStarted()
+        {
+            return (msgFeeder != null);
+        }
+
+        internal void enableAsyncProcessing()
+        {
+            if (msgFeeder == null)
+            {
+                msgFeeder = new Task(() => { conn.deliverMsgs(mch); });
+                msgFeeder.Start();
+            }
+        }
+
+        internal void disableAsyncProcessing()
+        {
+            if (msgFeeder != null)
+            {
+                mch.close();               
+                msgFeeder = null;
+            }
+        }
+
+        /// <summary>
+        /// Adds or removes a message handler to this subscriber.
+        /// </summary>
+        /// <remarks><see cref="MsgHandler">See MsgHandler</see></remarks>
+        public event MsgHandler MessageHandler
+        {
+            add
+            {
+                msgHandler += value;
+            }
+            remove
+            {
+                msgHandler -= value;
+            }
+        }
+
+        /// <summary>
+        /// This completes the subsciption process notifying the server this subscriber
+        /// has interest.
+        /// </summary>
+        public void Start()
+        {
+            if (isStarted())
+                return;
+
+            if (conn == null)
+                throw new NATSBadSubscriptionException();
+
+            conn.sendSubscriptonMessage(this);
+            enableAsyncProcessing();
+        }
+
+        override public void Unsubscribe()
+        {
+            disableAsyncProcessing();
+            base.Unsubscribe();
+        }
+
+        public override void AutoUnsubscribe(int max)
+        {
+            if (!isStarted())
+                Start();
+
+            base.AutoUnsubscribe(max);
+        }
+    }
+}

NATS/Channel.cs

@@ -0,0 +1,104 @@
+// Copyright 2015 Apcera Inc. All rights reserved.
+
+using System;
+using System.Collections.Generic;
+using System.Collections.Concurrent;
+using System.Linq;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace NATS.Client
+{
+    // Roll our own Channels - Concurrent Bag is more heavyweight
+    // than we need.
+    internal sealed class Channel<T>
+    {
+        Queue<T> q;
+        Object     qLock = new Object();
+        bool       finished = false;
+
+        internal Channel()
+        {
+            q = new Queue<T>(1024);
+        }
+
+        internal Channel(int initialCapacity)
+        {
+            q = new Queue<T>(initialCapacity);
+        }
+
+        internal T get(int timeout)
+        {
+            lock (qLock)
+            {
+                if (finished)
+                    return default(T);
+
+                if (q.Count > 0)
+                {
+                    return q.Dequeue();
+                }
+                else
+                {
+                    if (timeout < 0)
+                    {
+                        Monitor.Wait(qLock);
+                    }
+                    else
+                    {
+                        if (Monitor.Wait(qLock, timeout) == false)
+                        {
+                            throw new NATSTimeoutException();
+                        }
+                    }
+
+                    // we waited..
+                    if (finished)
+                        return default(T);
+
+                    return q.Dequeue();
+                }
+            }
+
+        } // get
+
+        internal void add(T item)
+        {
+            lock (qLock)
+            {
+                q.Enqueue(item);
+
+                // if the queue count was previously zero, we were
+                // waiting, so signal.
+                if (q.Count <= 1)
+                {
+                    Monitor.Pulse(qLock);
+                }
+            }
+        }
+
+        internal void close()
+        {
+            lock (qLock)
+            {
+                finished = true;
+                Monitor.Pulse(qLock);
+            }
+        }
+
+        internal int Count
+        {
+            get
+            {
+                lock (qLock)
+                {
+                    return q.Count;
+                }
+            }
+        }
+
+    } // class Channel
+
+}
+