代码之家  ›  专栏  ›  技术社区  ›  Johnny

如何从p2p中检索消息列表

  •  0
  • Johnny  · 技术社区  · 16 年前

    我有一个使用p2p的消息传递系统,每个对等端都有一个传入消息列表和一个传出消息列表。 我需要做的是,每当一个新的对等方加入网格时,他将从其他对等方获取所有传入消息,并将这些消息添加到自己的传入消息列表中。现在我知道了,当我从其他同事那里得到信息时,我可以让他们把自己的列表给我。但我找不到路怎么走…? 任何关于这方面的建议或帮助都将受到高度赞赏。我在下面给出我的代码。

       #region Instance Fields
    
        private string strOrigin = "";
        //the chat member name
        private string m_Member;
        //the channel instance where we execute our service methods against
        private IServerChannel m_participant;
        //the instance context which in this case is our window since it is the service host
        private InstanceContext m_site;
        //our binding transport for the p2p mesh
        private NetPeerTcpBinding m_binding;
        //the factory to create our chat channel
        private ChannelFactory<IServerChannel> m_channelFactory;
        //an interface provided by the channel exposing events to indicate
        //when we have connected or disconnected from the mesh
        private IOnlineStatus o_statusHandler;
        //a generic delegate to execute a thread against that accepts no args
        private delegate void NoArgDelegate();
        //an object to hold user details
        private IUserService userService;        
        //an Observable Collection of object to get all the Application Instance Details in databas
        ObservableCollection<AppLoginInstance> appLoginInstances;
        // an Observable Collection of object to get all Incoming Messages types
        ObservableCollection<MessageType> inComingMessageTypes;
        // an Observable Collection of object to get all Outgoing Messages
        ObservableCollection<PDCL.ERP.DataModels.Message> outGoingMessages;
        // an Observable Collection of object to get all Incoming Messages
        ObservableCollection<PDCL.ERP.DataModels.Message> inComingMessages;
        //an Event Aggregator to publish event for other modules to subscribe
        private readonly IEventAggregator eventAggregator;   
    
        /// <summary>
        /// an IUnityCOntainer to get the container
        /// </summary>
        private IUnityContainer container;
        private RefreshConnectionStatus refreshConnectionStatus;
        private RefreshConnectionStatusEventArgs args;
        private ReplyRequestMessage replyMessageRequest;
        private ReplyRequestMessageEventArgs eventsArgs;
    
        #endregion
    
        public P2pMessageService(IUserService UserService, IEventAggregator EventAggregator, IUnityContainer container)
        {
            userService = UserService;
            this.container = container;
            appLoginInstances = new ObservableCollection<AppLoginInstance>();
            inComingMessageTypes = new ObservableCollection<MessageType>();
            inComingMessages = new ObservableCollection<PDCL.ERP.DataModels.Message>();
            outGoingMessages = new ObservableCollection<PDCL.ERP.DataModels.Message>();
            this.args = new RefreshConnectionStatusEventArgs();
            this.eventsArgs = new ReplyRequestMessageEventArgs();
            this.eventAggregator = EventAggregator;            
            this.refreshConnectionStatus = this.eventAggregator.GetEvent<RefreshConnectionStatus>();
            this.replyMessageRequest = this.eventAggregator.GetEvent<ReplyRequestMessage>();
        }    
    
    
        #region IOnlineStatus Event Handlers
        void ostat_Offline(object sender, EventArgs e)
        {
            // we could update a status bar or animate an icon to 
            //indicate to the user they have disconnected from the mesh
    
            //currently i don't have a "disconnect" button but adding it
            //should be trivial if you understand the rest of this code
        }
    
        void ostat_Online(object sender, EventArgs e)
        {
            try
            {                
                m_participant.Join(userService.AppInstance);
            }
            catch (Exception Ex)
            {
                Logger.Exception(Ex, Ex.TargetSite.Name + ": " + Ex.TargetSite + ": " + Ex.Message);
            }
        }
    
        #endregion        
    
        #region IServer Members
    
        //this method gets called from a background thread to 
        //connect the service client to the p2p mesh specified
        //by the binding info in the app.config
        public void ConnectToMesh()
        {
            try
            {               
    
                m_site = new InstanceContext(this);
    
                //use the binding from the app.config with default settings
                m_binding = new NetPeerTcpBinding("P2PMessageBinding");
    
                m_channelFactory = new DuplexChannelFactory<IServerChannel>(m_site, "P2PMessageEndPoint");
                m_participant = m_channelFactory.CreateChannel();
                o_statusHandler = m_participant.GetProperty<IOnlineStatus>();
                o_statusHandler.Online += new EventHandler(ostat_Online);
                o_statusHandler.Offline += new EventHandler(ostat_Offline);
                //m_participant.InitializeMesh();
                //this.appLoginInstances.Add(this.userService.AppInstance);
    
                BackgroundWorkerHelper.DoWork<object>(() =>
                {
                    //this is an empty unhandled method on the service interface.
                    //why? because for some reason p2p clients don't try to connect to the mesh
                    //until the first service method call.  so to facilitate connecting i call this method
                    //to get the ball rolling.
                    m_participant.InitializeMesh();
                    //SynchronizeMessage(this.inComingMessages);    
                    return new object();
                }, arg =>
                {
    
                });
    
                this.appLoginInstances.Add(this.userService.AppInstance);
    
            }
            catch (Exception Ex)
            {
                Logger.Exception(Ex, Ex.TargetSite.Name + ": " + Ex.TargetSite + ": " + Ex.Message);
            }
        }
    
    
        public void Join(AppLoginInstance obj)
        {
            try
            {
                // Adding Instance to the PeerList
    
                if (appLoginInstances.SingleOrDefault(a => a.InstanceId == obj.InstanceId)==null)
                {
                    appLoginInstances.Add(obj);                    
                    this.refreshConnectionStatus.Publish(new RefreshConnectionStatusEventArgs() { Status = m_channelFactory.State });                                        
                }                    
    
                //this will retrieve any new members that have joined before the current user
                m_participant.SynchronizeMemberList(userService.AppInstance);
    
            }
            catch(Exception Ex)
            {
                Logger.Exception(Ex,Ex.TargetSite.Name + ": " + Ex.TargetSite + ": " + Ex.Message);
            }
        }
    
        /// <summary>
        /// Synchronizes member list
        /// </summary>
        /// <param name="obj">The AppLoginInstance Param</param>
        public void SynchronizeMemberList(AppLoginInstance obj)
        {
    
            //as member names come in we simply disregard duplicates and 
            //add them to the member list, this way we can retrieve a list
            //of members already in the chatroom when we enter at any time.
    
            //again, since this is just an example this is the simplified
            //way to do things.  the correct way would be to retrieve a list
            //of peernames and retrieve the metadata from each one which would
            //tell us what the member name is and add it.  we would want to check
            //this list when we join the mesh to make sure our member name doesn't 
            //conflict with someone else
            try
            {
                if (appLoginInstances.SingleOrDefault(a => a.InstanceId == obj.InstanceId) == null)
                {
                    appLoginInstances.Add(obj);                    
                }
            }
            catch (Exception Ex)
            {
                Logger.Exception(Ex, Ex.TargetSite.Name + ": " + Ex.TargetSite + ": " + Ex.Message);
            }
        }      
    
        /// <summary>
        /// This methos broadcasts the mesasge to all peers.
        /// </summary>
        /// <param name="msg">The whole message which is to be broadcasted</param>
        /// <param name="securityLevels"> Level of security</param>
        public void BroadCastMsg(PDCL.ERP.DataModels.Message msg, List<string> securityLevels)
        {
            try
            {
                foreach (string s in securityLevels)
                {
                    if (this.userService.IsInRole(s))
                    {
                        if (this.inComingMessages.Count == 0 && msg.CreatedByApp != this.userService.AppInstanceId)
                        {
                            this.inComingMessages.Add(msg);                            
                        }
                        else if (this.inComingMessages.SingleOrDefault(a => a.MessageId == msg.MessageId) == null && msg.CreatedByApp != this.userService.AppInstanceId)
                        {
                            this.inComingMessages.Add(msg);
                        }
                    }
                }
            }
            catch (Exception Ex)
            {
                Logger.Exception(Ex, Ex.TargetSite.Name + ": " + Ex.TargetSite + ": " + Ex.Message);
            }
    
        }
    
        /// <summary>
        /// 
        /// </summary>
        /// <param name="msg">The Message to denyed</param>
        public void BroadCastReplyMsg(PDCL.ERP.DataModels.Message msg)
        {
            try
            {
                //if (this.inComingMessages.SingleOrDefault(a => a.MessageId == msg.MessageId) != null)
                //{
                    this.replyMessageRequest.Publish(new ReplyRequestMessageEventArgs() { Message = msg });
                    this.inComingMessages.Remove(this.inComingMessages.SingleOrDefault(o => o.MessageId == msg.MessageId));
                //}
    
            }
            catch (Exception ex)
            {
                Logger.Exception(ex, ex.TargetSite.Name + ": " + ex.TargetSite + ": " + ex.Message);
            }
        }
    
        //again we need to sync the worker thread with the UI thread via Dispatcher
        public void Whisper(string Member, string MemberTo, string Message)
        {          
    
    
        }
    
        public void InitializeMesh()
        {
            //do nothing
        }
    
        public void Leave(AppLoginInstance obj)
        {
            if (this.appLoginInstances.SingleOrDefault(a => a.InstanceId == obj.InstanceId) != null)
            {
                this.appLoginInstances.Remove(this.appLoginInstances.Single(a => a.InstanceId == obj.InstanceId));
            }            
        }        
    
        //public void SynchronizeRemoveMemberList(AppLoginInstance obj)
        //{
        //    if (appLoginInstances.SingleOrDefault(a => a.InstanceId == obj.InstanceId) != null)
        //    {
        //        appLoginInstances.Remove(obj);
        //    }
        //}       
    
        #endregion
    
    1 回复  |  直到 8 年前
        1
  •  0
  •   Matthew Steeples    16 年前

    WCF中的P2P协议本质上是一种广播协议。将消息发送到特定对等端的唯一方法是获取其地址并通过不同的绑定与之交谈。问题在于,如果一个对等连接,那么所有当前节点都将希望连接到它来更新列表。

    您必须在一个单独的绑定上创建一个单独的方法,以便客户机可以检索列表。


    有关更多信息,请访问 PeerChannel Blog 可以找到特定于同步的信息 here . 此方法主要取决于网格的大小、消息的大小和消息的内容。

    例如,如果您的消息相对较小并且具有唯一的标识符,那么您可以在网格上以一个跃点计数重新传输消息,这意味着某些节点将重新接收消息。

    或者,您可以发送一条消息,其中包含包含包含大量消息的节点的IP地址,并使用另一个WCF连接(如basichttpbinding)到网格,然后任何需要消息的计算机都可以连接到其中一个节点。