P2P编程

开发平台：
Delphi

P2pServer.pas：源码内容
							Unit P2PServer;
Interface
Uses
  Windows, Messages, SysUtils, Classes,
  StdCtrls, IniFiles,
  P2pConst, P2pCommon,
  IdUDPServer, IdSocketHandle,
  NMUDP;
Type
  TP2PServer = Class;
  {检查存活的线程}
  TCheckLivingThread = Class(TThread)
  Public
    CheckTime: Cardinal;
    Server: TP2PServer;
    Procedure Execute; Override;
    Constructor Create(IStop: boolean; IServer: TP2PServer);
  End;
  TP2PServer = Class
  Private
    P2pLog: TMemo;
    FUserId: Cardinal;
    FBindings: TIDSocketHandles;
    Function GetUserId: Cardinal;
    Procedure SetBindings(Value: TIDSocketHandles);
  Protected
    Procedure Log(Const LogStr: String);
    Procedure OnUDPRead(Sender: TObject; AData: TStream; ABinding: TIdSocketHandle);
    Procedure OnP2pLogin(Data: Pointer; ABinding: TP2pBinding);
    Procedure OnP2pLoginOut(Data: Pointer; ABinding: TP2pBinding);
    Procedure OnP2pWantDoor(Data: Pointer; ABinding: TP2pBinding);
    Procedure OnP2pDoorOpened(Data: Pointer; ABinding: TP2pBinding);
    Procedure OnP2pAnyLifeing(Data: Pointer; ABinding: TP2pBinding);
    //广播事件
    Procedure GiveAllUserNewUser(IUser: TObject);
    Procedure GiveAllUserLeaveUser(IUser: TObject);
    // 发送事件
    Procedure GiveUserList(IUser: TObject);
    Procedure DisUserConn(IUser: TObject);
    Property UserID: Cardinal Read GetUserId;
  Public
    UserList: THashedStringList; {用户列表}
    UDPSocket: TIdUDPServer;
    CheckLive: TCheckLivingThread; {检查线程}
    Constructor Create(Logs: TMemo); Overload;
    Destructor Destroy; Override;
    Function StartServer: boolean;
    Function StopServer: boolean;
    Property Bindings: TIDSocketHandles Read FBindings Write SetBindings;
  End;
Implementation
{ TCheckLivingThread }
Constructor TCheckLivingThread.Create(IStop: boolean; IServer: TP2PServer);
Begin
  Inherited Create(IStop);
  Server := IServer;
  FreeOnTerminate := True;
  CheckTime := GetTickCount;
End;
Procedure TCheckLivingThread.Execute;
Var
  i: Integer;
Begin
  While Not Terminated Do
  Begin
    If GetTickCount - CheckTime > 60 * 1000 Then
    Begin
      For i := 0 To Server.UserList.Count - 1 Do
      Begin // Iterate
        Try
          If Not Terminated Then
          Begin
            If abs(GetTickCount - TP2pUser(Server.UserList.Objects[i]).MMCfg.LastLiveTime) > abs(TP2pUser(Server.UserList.Objects[i]).MMCfg.Livinginv) Then
            Begin
              TP2pUser(Server.UserList.Objects[i]).MMCfg.LastLiveTime := GetTickCount;
              TP2pUser(Server.UserList.Objects[i]).MMCfg.LifePoint := TP2pUser(Server.UserList.Objects[i]).MMCfg.LifePoint - 1;
              If TP2pUser(Server.UserList.Objects[i]).MMCfg.LifePoint = 0 Then
                Server.DisUserConn(Server.UserList.Objects[i]);
            End;
          End;
        Except
          Continue;
        End;
      End; // for
      CheckTime := GetTickCount;
    End
    Else
      Sleep(300);
  End; // while
End;
{P2pServer}
Constructor TP2PServer.Create(Logs: TMemo);
Begin
  P2pLog := Logs;
  UDPSocket := TIdUDPServer.Create(Nil);
  FBindings := UDPSocket.Bindings;
End;
Destructor TP2PServer.Destroy;
Begin
  If UDPSocket.Active Then StopServer;
  If UDPSocket <> Nil Then FreeAndNil(UDPSocket);
  Inherited;
End;
{启动服务器}
Function TP2PServer.StartServer: boolean;
Begin
  Result := false;
  If UDPSocket.Active Then Exit;
  Log('正在启动服务...');
  FUserId := 0;
  UDPSocket.OnUDPRead := OnUDPRead;
  Try
    UDPSocket.Active := True;
    UserList := THashedStringList.Create;
    CheckLive := TCheckLivingThread.Create(false, Self);
    Log('服务启动成功.');
    Result := True;
  Except
    Log('服务启动失败.');
    Exit;
  End;
End;
{停止服务器}
Function TP2PServer.StopServer: boolean;
Var
  i: Integer;
Begin
  Result := false;
  Try
    If Not UDPSocket.Active Then Exit;
    CheckLive.Terminate;
    WaitForSingleObject(CheckLive.Handle, 500);
    If UserList <> Nil Then
    Begin
      For i := 0 To UserList.Count - 1 Do
        UserList.Objects[i].Free;
      FreeAndNil(UserList);
    End;
    UDPSocket.Active := false;
    Log('服务停止成功.');
    Result := True;
  Except
    Log('服务停止失败.');
    Exit;
  End;
End;
Procedure TP2PServer.SetBindings(Value: TIDSocketHandles);
Begin
  UDPSocket.Bindings := Value;
  FBindings := Value;
End;
{超时退出}
Procedure TP2PServer.DisUserConn(IUser: TObject);
Var
  Idx: Integer;
Begin
  Log(Format('用户%s超时断开系统.', [TP2pUser(IUser).Address]));
  Idx := UserList.IndexOf(Format('%d', [TP2pUser(IUser).ID]));
  If Idx < 0 Then Exit;
  GiveAllUserLeaveUser(UserList.Objects[Idx]); {广播用户退出}
  UserList.Objects[Idx].Free;
  UserList.Delete(Idx);
End;
Function TP2PServer.GetUserId: Cardinal;
Begin
  Inc(FUserId);
  Result := FUserId;
End;
{日志事件}
Procedure TP2PServer.Log(Const LogStr: String);
Begin
  P2pLog.Lines.Add(Format('%s %s', [FormatDateTime('YYYY-MM-DD HH:NN:SS', Now), LogStr]));
  SendMessage(P2pLog.Handle, EM_SCROLLCARET, 0, 0);
End;
Procedure TP2PServer.OnUDPRead(Sender: TObject; AData: TStream; ABinding: TIdSocketHandle);
Var
  Ptr: Pointer;
  i: Integer;
  Binding: TP2pBinding;
Begin
  Try
    Ptr := TMemoryStream(AData).Memory;
    i := PInt(Ptr)^;
    Binding.PeerIP := ABinding.PeerIP;
    Binding.PeerPort := ABinding.PeerPort;
    Log(Format('接收到用户命令：<%d>.', [i]));
    Case i Of
      P2pLogin: OnP2pLogin(Ptr, Binding); //登陆服务器
      P2pLoginOut: OnP2pLoginOut(Ptr, Binding); //退出服务器
      P2pWantDoor: OnP2pWantDoor(Ptr, Binding); //用户请求开门
      P2pDoorOpened: OnP2pDoorOpened(Ptr, Binding); //用户通知门已经打开了
      P2pAnyLifeing: OnP2pAnyLifeing(Ptr, Binding); //发送心跳包
    End;
  Except
    On e: Exception Do
      Log(Format('处理用户数据时异常原因<Class%s -> %s>', [e.ClassName, e.Message]));
  End;
End;
{登陆服务器}
Procedure TP2PServer.OnP2pLogin(Data: Pointer; ABinding: TP2pBinding);
Var
  Rlb: TResutLoginBack;
  User: TP2pUser;
Begin
  Log(Format('来自%s:%d的用户发起登陆信息.', [ABinding.PeerIP, ABinding.PeerPort]));
  Rlb.ID.ID := P2pResultLoginBack;
  User := TP2pUser.Create(ABinding);
  User.ID := UserID;
  Rlb.UserID := User.ID;
  Log(Format('用户[%d]登陆成功.', [User.ID]));
  User.Call := PLoginUser(Data)^.Name;
  Rlb.State := 1;
  Rlb.Addr := Format('%s:%d', [ABinding.PeerIP, ABinding.PeerPort]);
  User.Socket.SendBuffer(Rlb, SizeOf(Rlb)); // 返回登陆成功
  GiveAllUserNewUser(User); // 广播新用户加入（不包括自己）
  GiveUserList(User); // 给用户列表
  UserList.AddObject(Format('%d', [User.ID]), User);
End;
{退出服务器}
Procedure TP2PServer.OnP2pLoginOut(Data: Pointer; ABinding: TP2pBinding);
Var
  Idx: Integer;
Begin
  Idx := UserList.IndexOf(IntToStr(PLoginoutUser(Data)^.UserID));
  If Idx < 0 Then Exit;
  Log(Format('用户%s退出系统.', [TP2pUser(UserList.Objects[Idx]).Address]));
  GiveAllUserLeaveUser(UserList.Objects[Idx]); {广播用户退出}
  UserList.Objects[Idx].Free;
  UserList.Delete(Idx);
End;
{请求开门}
Procedure TP2PServer.OnP2pWantDoor(Data: Pointer; ABinding: TP2pBinding);
Var
  Source, Dest: Cardinal;
  Idx: Integer;
  SUser, DUser: TP2pUser;
  WD: TWantDoor;
Begin
  Source := PWantDoor(Data)^.SourceID;
  Idx := UserList.IndexOf(Format('%d', [Source]));
  If Idx < 0 Then Exit;
  SUser := TP2pUser(UserList.Objects[Idx]);
  Dest := PWantDoor(Data)^.DestID;
  Idx := UserList.IndexOf(IntToStr(Dest));
  If Idx = -1 Then Exit;
  DUser := TP2pUser(UserList.Objects[Idx]);
  Log(Format('用户%s申请P2P目标%s', [SUser.Address, DUser.Address]));
  WD.ID.ID := P2pSomeOneWantDoor;
  WD.SourceID := Source;
  DUser.Socket.SendBuffer(WD, SizeOf(WD));
End;
{门开了}
Procedure TP2PServer.OnP2pDoorOpened(Data: Pointer; ABinding: TP2pBinding);
Var
  Idx: Integer;
  WD: TWantDoor;
Begin
  Idx := UserList.IndexOf(IntToStr(PWantDoor(Data)^.DestID));
  If Idx < 0 Then Exit;
  WD.ID.ID := P2pDoneDoor;
  Log(Format('用户%d已经为%s打开门.', [PWantDoor(Data)^.SourceID, TP2pUser(UserList.Objects[Idx]).Address]));
  WD.DestID := PWantDoor(Data)^.SourceID;
  TP2pUser(UserList.Objects[Idx]).Socket.SendBuffer(WD, SizeOf(WD));
End;
{心跳}
Procedure TP2PServer.OnP2pAnyLifeing(Data: Pointer; ABinding: TP2pBinding);
Var
  Idx: Integer;
  L: TLiving;
Begin
  Idx := UserList.IndexOf(Format('%d', [PLiving(Data)^.SourceUserId]));
  If Idx < 0 Then Exit;
  Log(Format('接收到%s:%d发过来的心跳包 名称:%s', [ABinding.PeerIP, ABinding.PeerPort, TP2pUser(UserList.Objects[Idx]).Address]));
  TP2pUser(UserList.Objects[Idx]).MMCfg.LifePoint := 2;
  TP2pUser(UserList.Objects[Idx]).MMCfg.LastLiveTime := GetTickCount;
  //回答心跳
  L.ID.ID := P2pAnyToAntLifeing;
  L.SourceUserId := 0;
  TP2pUser(UserList.Objects[Idx]).Socket.SendBuffer(L, SizeOf(L));
End;
{广播有新用户进入}
Procedure TP2PServer.GiveAllUserNewUser(IUser: TObject);
Var
  i: Integer;
  NewUser: TNewUser;
Begin
  With NewUser Do
  Begin
    ID.ID := P2PALLNewUser;
    Ip := TP2pUser(IUser).Binding.PeerIP;
    Port := TP2pUser(IUser).Binding.PeerPort;
    Call := TP2pUser(IUser).Call;
    UserID := TP2pUser(IUser).ID;
    For i := 0 To UserList.Count - 1 Do
      TP2pUser(UserList.Objects[i]).Socket.SendBuffer(NewUser, SizeOf(NewUser));
  End;
End;
{广播有用户离开}
Procedure TP2PServer.GiveAllUserLeaveUser(IUser: TObject);
Var
  i: Integer;
  NewUser: TLoginoutUser;
Begin
  NewUser.ID.ID := P2PALLLeaveUser;
  NewUser.UserID := TP2pUser(IUser).ID;
  For i := 0 To UserList.Count - 1 Do
    TP2pUser(UserList.Objects[i]).Socket.SendBuffer(NewUser, SizeOf(NewUser));
End;
{发送所有列表}
Procedure TP2PServer.GiveUserList(IUser: TObject);
Var
  i: Integer;
  MS: TMemoryStream;
  Head: Thead;
  Count: Word;
  Len: Word;
  Item: String;
Begin
  Log(Format('向用户%s发送用户列表.', [TP2pUser(IUser).Address]));
  MS := TMemoryStream.Create;
  Try
    Head.ID := P2PGiveUserlist;
    Count := UserList.Count;
    For i := 0 To Count - 1 Do
    Begin
      If i + 1 < Count Then
        Item := Format('%s%s:%d:%s' + #$D#$A, [Item, TP2pUser(UserList.Objects[i]).Address, TP2pUser(UserList.Objects[i]).ID, TP2pUser(UserList.Objects[i]).Call])
      Else
        Item := Format('%s%s:%d:%s', [Item, TP2pUser(UserList.Objects[i]).Address, TP2pUser(UserList.Objects[i]).ID, TP2pUser(UserList.Objects[i]).Call])
    End;
    Len := Length(Item);
    MS.SetSize(SizeOf(Head) + SizeOf(Count) + Len);
    MS.WriteBuffer(Head, SizeOf(Head));
    MS.WriteBuffer(Count, SizeOf(Count));
    MS.WriteBuffer(Len, SizeOf(Len));
    MS.WriteBuffer(Pointer(Item)^, Len);
    TP2pUser(IUser).Socket.SendStream(MS);
  Finally
    MS.Free;
  End;
End;
End.

						