WCF Duplex Callback Sample failing

Question

In an effort to hone some example services to be used as reference for our internal scenarios, I've created this WCF Duplex Channel example, pulling together several examples found through the years.

The duplex part isn't working and I'm hoping we can all figure it out together. I hate posting this much code, but I feel I've trimmed this down as short as WCF can go, while incorporating all the parts I'm hoping to have vetted by the community. There might be some really bad ideas in here, I'm not saying it's right, it's just what I've got so far.

There are three parts. The Channel, the Server, and the Client. Three projects, and here, three code files. No XML configuration, everything is coded in. Followed by the code output.

Channel.proj / Channel.cs

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.ServiceModel;

namespace Channel
{
    public interface IDuplexSyncCallback
    {
        [OperationContract]
        string CallbackSync(string message, DateTimeOffset timestamp);
    }

    [ServiceContract(CallbackContract = typeof(IDuplexSyncCallback))]
    public interface IDuplexSyncContract
    {
        [OperationContract]
        void Ping();

        [OperationContract]
        void Enroll();

        [OperationContract]
        void Unenroll();
    }
}

Server.proj / Server.cs, references Channel, System.Runtime.Serialization, System.ServiceModel

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.ServiceModel;
using System.Timers;
using Channel;
using System.Diagnostics;
using System.Net.Security;

namespace Server
{
    class Program
    {
        // All of this just starts up the service with these hardcoded configurations
        static void Main(string[] args)
        {
            ServiceImplementation implementation = new ServiceImplementation();
            ServiceHost service = new ServiceHost(implementation);

            NetTcpBinding binding = new NetTcpBinding(SecurityMode.Transport);
            binding.Security.Message.ClientCredentialType = MessageCredentialType.Windows;
            binding.Security.Mode = SecurityMode.Transport;
            binding.Security.Transport.ClientCredentialType = TcpClientCredentialType.Windows;
            binding.Security.Transport.ProtectionLevel = ProtectionLevel.EncryptAndSign;
            binding.ListenBacklog = 1000;
            binding.MaxConnections = 30;
            binding.MaxReceivedMessageSize = 2147483647;
            binding.ReaderQuotas.MaxStringContentLength = 2147483647;
            binding.ReaderQuotas.MaxArrayLength = 2147483647;
            binding.SendTimeout = TimeSpan.FromSeconds(2);
            binding.ReceiveTimeout = TimeSpan.FromSeconds(10 * 60); // 10 minutes is the default if not specified
            binding.ReliableSession.Enabled = true;
            binding.ReliableSession.Ordered = true;

            service.AddServiceEndpoint(typeof(IDuplexSyncContract), binding, new Uri("net.tcp://localhost:3828"));

            service.Open();

            Console.WriteLine("Server Running ... Press any key to quit");
            Console.ReadKey(true);

            service.Abort();
            service.Close();
            implementation = null;
            service = null;
        }
    }

    /// <summary>
    /// ServiceImplementation of IDuplexSyncContract
    /// </summary>
    [ServiceBehavior(InstanceContextMode = InstanceContextMode.Single,
        MaxItemsInObjectGraph = 2147483647,
        IncludeExceptionDetailInFaults = true,
        ConcurrencyMode = ConcurrencyMode.Multiple,
        UseSynchronizationContext = false)]
    class ServiceImplementation : IDuplexSyncContract
    {
        Timer announcementTimer = new Timer(5000); // Every 5 seconds
        int messageNumber = 0; // message number incrementer - not threadsafe, just for debugging.

        public ServiceImplementation()
        {
            announcementTimer.Elapsed += new ElapsedEventHandler(announcementTimer_Elapsed);
            announcementTimer.AutoReset = true;
            announcementTimer.Enabled = true;
        }

        void announcementTimer_Elapsed(object sender, ElapsedEventArgs e)
        {
            AnnounceSync(string.Format("HELLO? (#{0})", messageNumber++));
        }

        #region IDuplexSyncContract Members
        List<IDuplexSyncCallback> syncCallbacks = new List<IDuplexSyncCallback>();

        /// <summary>
        /// Simple Ping liveness
        /// </summary>
        [OperationBehavior]
        public void Ping() { return; }

        /// <summary>
        /// Add channel to subscribers
        /// </summary>
        [OperationBehavior]
        void IDuplexSyncContract.Enroll()
        {
            IDuplexSyncCallback current = System.ServiceModel.OperationContext.Current.GetCallbackChannel<IDuplexSyncCallback>();

            lock (syncCallbacks)
            {
                syncCallbacks.Add(current);

                Trace.WriteLine("Enrollment Complete");
            }
        }

        /// <summary>
        /// Remove channel from subscribers
        /// </summary>
        [OperationBehavior]
        void IDuplexSyncContract.Unenroll()
        {
            IDuplexSyncCallback current = System.ServiceModel.OperationContext.Current.GetCallbackChannel<IDuplexSyncCallback>();

            lock (syncCallbacks)
            {
                syncCallbacks.Remove(current);

                Trace.WriteLine("Unenrollment Complete");
            }
        }

        /// <summary>
        /// Callback to clients over enrolled channels
        /// </summary>
        /// <param name="message"></param>
        void AnnounceSync(string message)
        {
            var now = DateTimeOffset.Now;

            if (message.Length > 2000) message = message.Substring(0, 2000 - "[TRUNCATED]".Length) + "[TRUNCATED]";
            Trace.WriteLine(string.Format("{0}: {1}", now.ToString("mm:ss.fff"), message));

            lock (syncCallbacks)
            {
                foreach (var callback in syncCallbacks.ToArray())
                {
                    Console.WriteLine("Sending \"{0}\" synchronously ...", message);

                    CommunicationState state = ((ICommunicationObject)callback).State;

                    switch (state)
                    {
                        case CommunicationState.Opened:
                            try
                            {
                                Console.WriteLine("Client said '{0}'", callback.CallbackSync(message, now));
                            }
                            catch (Exception ex)
                            {
                                // Timeout Error happens here
                                syncCallbacks.Remove(callback);
                                Console.WriteLine("Removed client");
                            }
                            break;
                        case CommunicationState.Created:
                        case CommunicationState.Opening:
                            break;
                        case CommunicationState.Faulted:
                        case CommunicationState.Closed:
                        case CommunicationState.Closing:
                        default:
                            syncCallbacks.Remove(callback);
                            Console.WriteLine("Removed client");
                            break;
                    }
                }
            }
        }
        #endregion
    }
}

Client.proj / Client.cs, references Channel, System.Runtime.Serialization, System.ServiceModel

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.ServiceModel;
using System.Timers;
using System.Diagnostics;
using Channel;
using System.Net;

namespace Client
{
    class Program
    {
        static void Main(string[] args)
        {
            using (var callbackSyncProxy = new CallbackSyncProxy(new Uri("net.tcp://localhost:3828"), CredentialCache.DefaultNetworkCredentials))
            {
                callbackSyncProxy.Faulted += (s, e) => Console.WriteLine("CallbackSyncProxy Faulted.");
                callbackSyncProxy.ConnectionUnavailable += (s, e) => Console.WriteLine("CallbackSyncProxy ConnectionUnavailable.");
                callbackSyncProxy.ConnectionRecovered += (s, e) => Console.WriteLine("CallbackSyncProxy ConnectionRecovered.");

                callbackSyncProxy.Ping();
                callbackSyncProxy.Ping();
                callbackSyncProxy.Ping();

                Console.WriteLine("Pings completed.  Enrolling ...");

                callbackSyncProxy.AnnouncementSyncHandler = AnnouncementHandler;

                Console.WriteLine("Enrolled and waiting.  Press any key to quit ...");
                Console.ReadKey(true); // Wait for quit
            }
        }

        /// <summary>
        /// Called by the server through DuplexChannel
        /// </summary>
        /// <param name="message"></param>
        /// <param name="timeStamp"></param>
        /// <returns></returns>
        static string AnnouncementHandler(string message, DateTimeOffset timeStamp)
        {
            Console.WriteLine("{0}: {1}", timeStamp, message);

            return string.Format("Dear Server, thanks for that message at {0}.", timeStamp);
        }
    }

    /// <summary>
    /// Encapsulates the client-side WCF setup logic.
    /// 
    /// There are 3 events Faulted, ConnectionUnavailable, ConnectionRecovered that might be of interest to the consumer
    /// Enroll and Unenroll of the ServiceContract are called when setting an AnnouncementSyncHandler
    /// Ping, when set correctly against the server's send/receive timeouts, will keep the connection alive
    /// </summary>
    public class CallbackSyncProxy : IDisposable
    {
        Uri listen;
        NetworkCredential credentials;
        NetTcpBinding binding;
        EndpointAddress serverEndpoint;
        ChannelFactory<IDuplexSyncContract> channelFactory;
        DisposableChannel<IDuplexSyncContract> channel;

        readonly DuplexSyncCallback callback = new DuplexSyncCallback();

        object sync = new object();
        bool enrolled;
        Timer pingTimer = new Timer();
        bool quit = false; // set during dispose

        // Events of interest to consumer
        public event EventHandler Faulted;
        public event EventHandler ConnectionUnavailable;
        public event EventHandler ConnectionRecovered;

        // AnnouncementSyncHandler property.  When set to non-null delegate, Enrolls client with server.
        // passes through to the DuplexSyncCallback callback.AnnouncementSyncHandler
        public Func<string, DateTimeOffset, string> AnnouncementSyncHandler
        {
            get
            {
                Func<string, DateTimeOffset, string> temp = null;

                lock (sync)
                {
                    temp = callback.AnnouncementSyncHandler;
                }
                return temp;
            }
            set
            {
                lock (sync)
                {
                    if (callback.AnnouncementSyncHandler == null && value != null)
                    {
                        callback.AnnouncementSyncHandler = value;

                        Enroll();
                    }
                    else if (callback.AnnouncementSyncHandler != null && value == null)
                    {
                        Unenroll();

                        callback.AnnouncementSyncHandler = null;
                    }
                    else // null to null or function to function, just update it
                    {
                        callback.AnnouncementSyncHandler = value;
                    }
                }
            }
        }

        /// <summary>
        /// using (var proxy = new CallbackSyncProxy(listen, CredentialCache.DefaultNetworkCredentials) { ... }
        /// </summary>
        public CallbackSyncProxy(Uri listen, NetworkCredential credentials)
        {
            this.listen = listen;
            this.credentials = credentials;

            binding = new NetTcpBinding(SecurityMode.Transport);
            binding.Security.Message.ClientCredentialType = MessageCredentialType.Windows;
            binding.Security.Mode = SecurityMode.Transport;
            binding.Security.Transport.ClientCredentialType = TcpClientCredentialType.Windows;
            binding.MaxReceivedMessageSize = 2147483647;
            binding.ReaderQuotas.MaxArrayLength = 2147483647;
            binding.ReaderQuotas.MaxBytesPerRead = 2147483647;
            binding.ReaderQuotas.MaxDepth = 2147483647;
            binding.ReaderQuotas.MaxStringContentLength = 2147483647;
            binding.ReliableSession.Enabled = true;
            binding.ReliableSession.Ordered = true;
            serverEndpoint = new EndpointAddress(listen);

            pingTimer.AutoReset = true;
            pingTimer.Elapsed += pingTimer_Elapsed;
            pingTimer.Interval = 20000;
        }

        /// <summary>
        /// Keep the connection alive by pinging at some set minimum interval
        /// </summary>
        void pingTimer_Elapsed(object sender, ElapsedEventArgs e)
        {
            bool locked = false;

            try
            {
                locked = System.Threading.Monitor.TryEnter(sync, 100);
                if (!locked)
                {
                    Console.WriteLine("Unable to ping because synchronization lock could not be aquired in a timely fashion");
                    return;
                }
                Debug.Assert(channel != null, "CallbackSyncProxy.channel is unexpectedly null");

                try
                {
                    channel.Service.Ping();
                }
                catch
                {
                    Console.WriteLine("Unable to ping");
                }
            }
            finally
            {
                if (locked) System.Threading.Monitor.Exit(sync);
            }
        }

        /// <summary>
        /// Ping is a keep-alive, but can also be called by the consuming code
        /// </summary>
        public void Ping()
        {
            lock (sync)
            {
                if (channel != null)
                {
                    channel.Service.Ping();
                }
                else
                {
                    using (var c = new DisposableChannel<IDuplexSyncContract>(GetChannelFactory().CreateChannel()))
                    {
                        c.Service.Ping();
                    }
                }
            }
        }

        /// <summary>
        /// Enrollment - called when AnnouncementSyncHandler is assigned
        /// </summary>
        void Enroll()
        {
            lock (sync)
            {
                if (!enrolled)
                {
                    Debug.Assert(channel == null, "CallbackSyncProxy.channel is unexpectedly not null");

                    var c = new DisposableChannel<IDuplexSyncContract>(GetChannelFactory().CreateChannel());

                    ((ICommunicationObject)c.Service).Open();

                    ((ICommunicationObject)c.Service).Faulted += new EventHandler(CallbackChannel_Faulted);

                    c.Service.Enroll();

                    channel = c;

                    Debug.Assert(!pingTimer.Enabled, "CallbackSyncProxy.pingTimer unexpectedly Enabled");

                    pingTimer.Start();

                    enrolled = true;
                }
            }
        }

        /// <summary>
        /// Unenrollment - called when AnnouncementSyncHandler is set to null
        /// </summary>
        void Unenroll()
        {
            lock (sync)
            {
                if (callback.AnnouncementSyncHandler != null)
                {
                    Debug.Assert(channel != null, "CallbackSyncProxy.channel is unexpectedly null");

                    channel.Service.Unenroll();

                    Debug.Assert(!pingTimer.Enabled, "CallbackSyncProxy.pingTimer unexpectedly Disabled");

                    pingTimer.Stop();

                    enrolled = false;
                }
            }
        }

        /// <summary>
        /// Used during enrollment to establish a channel.
        /// </summary>
        /// <returns></returns>
        ChannelFactory<IDuplexSyncContract> GetChannelFactory()
        {
            lock (sync)
            {
                if (channelFactory != null &&
                    channelFactory.State != CommunicationState.Opened)
                {
                    ResetChannel();
                }

                if (channelFactory == null)
                {
                    channelFactory = new DuplexChannelFactory<IDuplexSyncContract>(callback, binding, serverEndpoint);

                    channelFactory.Credentials.Windows.ClientCredential = credentials;

                    foreach (var op in channelFactory.Endpoint.Contract.Operations)
                    {
                        var b = op.Behaviors[typeof(System.ServiceModel.Description.DataContractSerializerOperationBehavior)] as System.ServiceModel.Description.DataContractSerializerOperationBehavior;

                        if (b != null)
                            b.MaxItemsInObjectGraph = 2147483647;
                    }
                }
            }

            return channelFactory;
        }

        /// <summary>
        /// Channel Fault handler, set during Enrollment
        /// </summary>
        void CallbackChannel_Faulted(object sender, EventArgs e)
        {
            lock (sync)
            {
                if (Faulted != null)
                {
                    Faulted(this, new EventArgs());
                }

                ResetChannel();

                pingTimer.Stop();
                enrolled = false;

                if (callback.AnnouncementSyncHandler != null)
                {
                    while (!quit) // set during Dispose
                    {
                        System.Threading.Thread.Sleep(500);

                        try
                        {
                            Enroll();

                            if (ConnectionRecovered != null)
                            {
                                ConnectionRecovered(this, new EventArgs());

                                break;
                            }
                        }
                        catch
                        {
                            if (ConnectionUnavailable != null)
                            {
                                ConnectionUnavailable(this, new EventArgs());
                            }
                        }
                    }
                }
            }
        }

        /// <summary>
        /// Reset the Channel & ChannelFactory if they are faulted and during dispose
        /// </summary>
        void ResetChannel()
        {
            lock (sync)
            {
                if (channel != null)
                {
                    channel.Dispose();
                    channel = null;
                }

                if (channelFactory != null)
                {
                    if (channelFactory.State == CommunicationState.Faulted)
                        channelFactory.Abort();
                    else
                        try
                        {
                            channelFactory.Close();
                        }
                        catch
                        {
                            channelFactory.Abort();
                        }

                    channelFactory = null;
                }
            }
        }

        // Disposing of me implies disposing of disposable members
        #region IDisposable Members
        bool disposed;
        void IDisposable.Dispose()
        {
            if (!disposed)
            {
                Dispose(true);
            }

            GC.SuppressFinalize(this);
        }

        void Dispose(bool disposing)
        {
            if (disposing)
            {
                quit = true;

                ResetChannel();

                pingTimer.Stop();

                enrolled = false;

                callback.AnnouncementSyncHandler = null;
            }

            disposed = true;
        }
        #endregion
    }

    /// <summary>
    /// IDuplexSyncCallback implementation, instantiated through the CallbackSyncProxy
    /// </summary>
    [CallbackBehavior(UseSynchronizationContext = false, 
    ConcurrencyMode = ConcurrencyMode.Multiple, 
    IncludeExceptionDetailInFaults = true)]
    class DuplexSyncCallback : IDuplexSyncCallback
    {
        // Passthrough handler delegates from the CallbackSyncProxy
        #region AnnouncementSyncHandler passthrough property
        Func<string, DateTimeOffset, string> announcementSyncHandler;
        public Func<string, DateTimeOffset, string> AnnouncementSyncHandler
        {
            get
            {
                return announcementSyncHandler;
            }
            set
            {
                announcementSyncHandler = value;
            }
        }
        #endregion

        /// <summary>
        /// IDuplexSyncCallback.CallbackSync
        /// </summary>
        [OperationBehavior]
        public string CallbackSync(string message, DateTimeOffset timestamp)
        {
            if (announcementSyncHandler != null)
            {
                return announcementSyncHandler(message, timestamp);
            }
            else
            {
                return "Sorry, nobody was home";
            }
        }
    }

    // This class wraps an ICommunicationObject so that it can be either Closed or Aborted properly with a using statement
    // This was chosen over alternatives of elaborate try-catch-finally blocks in every calling method, or implementing a
    // new Channel type that overrides Disposable with similar new behavior
    sealed class DisposableChannel<T> : IDisposable
    {
        T proxy;
        bool disposed;

        public DisposableChannel(T proxy)
        {
            if (!(proxy is ICommunicationObject)) throw new ArgumentException("object of type ICommunicationObject expected", "proxy");

            this.proxy = proxy;
        }

        public T Service
        {
            get
            {
                if (disposed) throw new ObjectDisposedException("DisposableProxy");

                return proxy;
            }
        }

        public void Dispose()
        {
            if (!disposed)
            {
                Dispose(true);
            }

            GC.SuppressFinalize(this);
        }

        void Dispose(bool disposing)
        {
            if (disposing)
            {
                if (proxy != null)
                {
                    ICommunicationObject ico = null;

                    if (proxy is ICommunicationObject)
                        ico = (ICommunicationObject)proxy;

                    // This state may change after the test and there's no known way to synchronize
                    // so that's why we just give it our best shot
                    if (ico.State == CommunicationState.Faulted)
                        ico.Abort(); // Known to be faulted
                    else
                        try
                        {
                            ico.Close(); // Attempt to close, this is the nice way and we ought to be nice
                        }
                        catch
                        {
                            ico.Abort(); // Sometimes being nice isn't an option
                        }

                    proxy = default(T);
                }
            }

            disposed = true;
        }
    }
}

Collated Output:

>> Server Running ... Press any key to quit
                           Pings completed.  Enrolling ... <<
          Enrolled and waiting.  Press any key to quit ... <<
>> Sending "HELLO? (#0)" synchronously ...
                                CallbackSyncProxy Faulted. <<
                    CallbackSyncProxy ConnectionRecovered. <<
>> Removed client
>> Sending "HELLO? (#2)" synchronously ...
                   8/2/2010 2:47:32 PM -07:00: HELLO? (#2) <<
>> Removed client

As Andrew has pointed out, the problem isn't so self-evident. This "collated output" is not the desired output. Instead, I would want the Server to be running, the Pings and enrollment to succeed, and then every 5 seconds, the server would "Sending "HELLO? (#m)" synchronously" and immediately the Client would transform and return and that the Server would receive and print out.

Instead, the pings work, but the Callback faults on the first try, gets to the Client on the reconnect but doesn't return to the Server, and everything disconnects.

The only exceptions I get to see relate to the channel having faulted previously and hence being unusable, but nothing yet on the actual fault that causes the channel to reach that state.

I've used similar code with [OperationalBehavior(IsOneWay= true)] plenty of times. Strange that this seemingly more common case is giving me such grief.

The exception caught on the server side, which I don't understand, is:
System.TimeoutException: "This request operation sent to schemas.microsoft.com/2005/12/ServiceModel/Addressing/Anonymous did not receive a reply within the configured timeout (00:00:00). The time allotted to this operation may have been a portion of a longer timeout. This may be because the service is still processing the operation or because the service was unable to send a reply message. Please consider increasing the operation timeout (by casting the channel/proxy to IContextChannel and setting the OperationTimeout property) and ensure that the service is able to connect to the client."

Answer 1

On a server in AnnounceSync method add FaultException handling and you would be informed that there is no response from server (which in your case is a client) meaning there is no callback received.

It is as you suggested due to timeout. So change

binding.SendTimeout = TimeSpan.FromSeconds(3);

It would work as expected.

try
{
    Console.WriteLine("Client said '{0}'",callback.CallbackSync(message, now) );
}
catch (FaultException fex)
{
    syncCallbacks.Remove(callback);
    Console.WriteLine("Failed to call Client because" + fex.Reason);
    Console.WriteLine(fex.Message);
}

Answer 2

It's very silly/aggravating, but it seems that the ProtectionLevel.EncryptAndSign is the problem. I found the error message on Google infrequently related to the bindings and Windows auth. Lead me to guess that perhaps the upstream communication was not working due to something related to the binding encryption ... or something. But setting it to ProtectionLevel.None instead suddenly allows the duplex channel to work for two-way methods (methods that return values back to the server)

I'm not saying that turning off the protection level is a good idea, but at least it is a significant lead. If you need the benefits of the EncryptAndSign, you can further investigate from there.

Answer 3

Just my $0.02; download the WCF & WF samples pack and use the Duplex sample instead. http://www.microsoft.com/downloads/details.aspx?FamilyID=35ec8682-d5fd-4bc3-a51a-d8ad115a8792&displaylang=en

Answer 4

Unfortunately, OneWay operations are a prerequisite for duplex channels.

Answer 5

This may not completely fix your issue but looking at your code, IDuplexSyncCallback is definitely a suspect. Part of it's service implementation is in place but it should be decorated with the ServiceContractAttribute as well. When doing a callback it must be designated as one-way as well. Below is a sample of what I have done in the past for a callback contract and may help you out as well.

[ServiceContract]
public interface IDuplexSyncCallback
{
    [OperationContract(IsOneWay = true)
    string CallbackSync(string message, DateTimeOffset timestamp);
}