WCF Publish/Subscribe and using callbacks to send data to specific users

Question

Hello thanks for looking,

I'm working on a project at the moment and have become a little stuck. I'm creating a client server app, which allows a client to subscribe to the server to have messages forwarded to it.

The issue I'm having is that when the client subscribes I wish for them to only recieve updates that relate to them. The system basically passes messages from a SQL server DB which the server monitors. When a new message is recieved the server should only forward the message to the clients that it applys to, based on whos logged on the client machine.

I've had a look and found code samples which sign up for messages to be broadcast across all clients who have subscribed, but not any that show how to identify individual clients and if messages apply to them.

If anyone could help or point me in the right direction it would be appreciated.

edit Just to be clearer I'm not so much wondering how to operate callbacks and subscription but how to operate a subscription service where when a user subscribes they can provide there user ID along with the callback information, which can then be used to identify which specific users, messages need to be sent to.

You can now find some of my code below:

namespace AnnouncementServiceLibrary
{
    [ServiceContract(CallbackContract = typeof(IMessageCallback))]
    public interface IMessageCheck
    {
        [OperationContract]
        void MessageCheck();
    }
}

namespace AnnouncementServiceLibrary
{
    public interface IMessageCallback
    {
        [OperationContract(IsOneWay = true)]
        void OnNewMessage(Mess message);
    }
}

Subscribe/Unsubscribe:

private static readonly List<IMessageCallback> subscribers = new List<IMessageCallback>();

        public bool Subscribe()
    {
        try
        {

            IMessageCallback callback = OperationContext.Current.GetCallbackChannel<IMessageCallback>();

            //If they dont already exist in the subscribers list, adds them to it
            if (!subscribers.Contains(callback))
                subscribers.Add(callback);
            return true;
        }
        catch
        {
            //Otherwise if an error occurs returns false
            return false;
        }
    }


    /// <summary>
    /// Unsubscribes the user from recieving new messages when they become avaliable
    /// </summary>
    /// <returns>Returns a bool that indicates whether the operation worked or not</returns>
    public bool Unsubscribe()
    {
        try
        {

            IMessageCallback callback = OperationContext.Current.GetCallbackChannel<IMessageCallback>();

            //If they exist in the list of subscribers they are then removed
            if (subscribers.Contains(callback))
                subscribers.Remove(callback);
            return true;
        }
        catch
        {
            //Otherwise if an error occurs returns false
            return false;
        }

    }

Finally this at the moment isnt't working as basically when a user subscribes as it loops through I want it to filter the LINQ query based on the users userID:

#region IMessageCheck Members

        /// <summary>
        /// This method checks for new messages recieved based on those who have subscribed for the service
        /// </summary>
        public void MessageCheck()
        {
            //A continuous loop to keep the method going
            while(true)
            {
                //Changes the thread to a sleep state for 2 mins?
                Thread.Sleep(200000);

                //Go through each subscriber based on there callback information
                subscribers.ForEach(delegate(IMessageCallback callback)
                {
                    //Checks if the person who wanted the callback can still be communicated with
                    if (((ICommunicationObject)callback).State == CommunicationState.Opened)
                    {
                        //Creates a link to the database and gets the required information
                        List<Mess> mess = new List<Mess>();
                        List<Message> me;
                        List<MessageLink> messLink;

                        AnnouncementDBDataContext aDb = new AnnouncementDBDataContext();

                        me = aDb.Messages.ToList();
                        messLink = aDb.MessageLinks.ToList();

                        //Query to retrieve any messages which are newer than the time when the last cycle finished
                        var result = (from a in messLink
                                      join b in me
                                          on a.UniqueID equals b.UniqueID
                                      where b.TimeRecieved > _time
                                      select new { b.UniqueID, b.Author, b.Title, b.Body, b.Priority, a.Read, b.TimeRecieved });

                        //Foreach result a new message is created and returned to the PC that subscribed
                        foreach (var a in result)
                        {
                            Mess message = new Mess(a.UniqueID, a.Author, a.Title, a.Body, a.Priority, (bool)a.Read, a.TimeRecieved);
                            callback.OnNewMessage(message);
                        }
                    }
                    //If the requesting PC can't be contacted they are removed from the subscribers list
                    else
                    {
                        subscribers.Remove(callback);
                    }
                });

                //Sets the datetime so the next cycle can measure against to see if new messages have been recieved
                _time = DateTime.Now;
            }

        }
        #endregion

Answer 1

You can use a DuplexChannel. For this, you have to provide a binding that supports both session communication and duplex communication. Then, the clients will have to pass a InstanceContext constructed with an instance of the CallbackHandler. Finally, the server will get the context (for callback messages), using:

OperationContext.Current.GetCallbackChannel<ICallBackServiceContract>();

where ICallBackServiceContract is the contract implemented on the client.

To learn more about Duplex Services, see: Duplex Services

EDIT: Well, if the callback is working fine, I mean it's a instance behavior. Try to add (or change) the contract implementation, using the PerSession instance mode:

[ServiceBehavior(InstanceContextMode=InstanceContextMode.PerSession)]

Answer 2

Have a look at Juval Lowy's Publish-Subscribe WCF Framework, which is described in pretty good detail in this MSDN article. The code is available to look at via the article, or you can download the source and example from Lowy's website here (you'll need to scroll back on the webpage "two scrollwheel turns" to see the links).

I am using this mechanism in my WCF application, and it works like a charm. Hope this helps.

Answer 3

There are many ways to accomplish this. Considering you are using a static List to maintain your subscribers you could generate a new object like so:

class Subscriber
{
    public string UserName { get; set; }
    public IMessageCallback CallBack { get; set; }
}

Then store your subscribers in a List<Subscriber> instead of a List<IMessageCallback> object.

You could then modify your Subscribe() method to take a string parameter for username. This would allow you to use your linq to objects query to find the user you want to send a message to.

This technique could work for any identifier, but I'm not sure how you are trying to filter the messages. Looks like you want it by username, that's why I used this option here. But you can just as easily have a flags Enum for they types of Subscriptions and pass that in.

If you want an alternative to storing your subscribers in a static List you can check out an article I wrote about Throttling WCF which I use GenericDelegates. This might give you more options and ideas. http://www.codeproject.com/KB/WCF/wcfesb.aspx . This article will also show you an easier way to maintain subscribers than checking context state on every call.