WCF双工回调示例失败

为了磨练一些示例服务,作为我们内部场景的参考,我创建了这个WCF双工通道示例,将多年来发现的几个示例放在一起。

双工部分坏了,我希望我们能一起解决。我讨厌发布这么多的代码,但我觉得我已经把它删减到了WCF所能做到的最短,同时合并了我希望社区能够审查的所有部分。这里可能有一些非常糟糕的想法,我不是说这是对的,这只是我到目前为止得到的。

Channel.proj/频道.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,引用通道,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,引用通道,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;
        }
    }
}

>> 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

正如安德鲁所指出的,这个问题并不是不言而喻的。此“整理输出”不是所需的输出。相反,我希望服务器运行,ping和注册成功,然后每隔5秒,服务器就会“发送”HELLO(#m) “同步”并且客户机将立即转换并返回,服务器将接收并打印输出。

相反,ping可以工作,但是第一次尝试时回调出错,在重新连接时到达客户机,但是没有返回到服务器,所有的东西都断开连接。

我看到的唯一例外与通道以前出现故障,因此无法使用有关,但与导致通道达到该状态的实际故障无关。

我用过类似的代码 [OperationalBehavior(IsOneWay= true)] 很多次。奇怪的是,这个看似更普通的案子却让我如此悲痛。

服务器端捕获的异常(我不明白)是:
System.TimeoutException:“发送到schemas.microsoft.com/2005/12/ServiceModel/Addressing/Anonymous的此请求操作在配置的超时(00:00:00)内未收到答复。分配给此操作的时间可能是较长超时的一部分。这可能是因为服务仍在处理该操作,或者因为服务无法发送回复消息。请考虑增加操作超时(通过将通道/代理强制转换为IContextChannel并设置OperationTimeout属性),并确保服务能够连接到客户端。“