RtspProxyServer.pas

unit RtspProxyServer;

interface

uses
  SysUtils, Classes, Windows, Sockets, RtspTypes, RtspProxyTube;

type
  TProxyClient = class;
  TServerThread = class;

  TProxyServer = class
  private
    FProxy: TTcpServer;
    FClients: TList;
    FClientsLocker: TRTLCriticalSection;
    procedure AddClient(AClient: TProxyClient);
    procedure RemoveClient(AClient: TProxyClient);
    procedure ClientAccepted(Sender: TObject; Client: TCustomIpClient);
  public
    constructor Create(LocalPort: TSocketPort);
    destructor Destroy; override;
    procedure Start;
    procedure Stop;
  end;

  TServerResponceEvent = procedure (Sender: TObject; Msg: TRtspMessage) of object;

  // Êëàññ îáñëóæèâàåò îäíî êëèåíòñêîå ïîäêëþ÷åíèå
  TProxyClient = class
  private
    FServer: TCustomIpClient;
    FClient: TCustomIpClient;
    FServerResource: String;
    FServerHost: TSocketHost;
    FServerPort: TSocketPort;
    FLocalResource: String;
    FLocalHost: TSocketHost;
    FLocalPort: TSocketPort;
    FProxyTube: TRtspProxyTube;
    FServerThread: TServerThread;
    procedure ConnectToServer(const Resource: String);
    procedure ProcessSetup1Responce(Sender: TObject; Msg: TRtspMessage);
    procedure ProcessSetup2Responce(Sender: TObject; Msg: TRtspMessage);
    procedure ProcessPlayResponce(Sender: TObject; Msg: TRtspMessage);
  public
    constructor Create(AClient: TCustomIpClient);
    destructor Destroy; override;
    procedure Listen;
    procedure Stop;
  end;

  TServerThread = class(TThread)
  private
    FOwner: TProxyClient;
    FClient: TCustomIpClient;
    FServer: TCustomIpClient;
    FOnResponce: TServerResponceEvent;
  public
    procedure Execute; override;
  end;

procedure LockProxy;
procedure UnlockProxy;

const
  BAD_ACTION_DISCARD = 1;
  BAD_ACTION_IGNORE = 2;

var
  BadClientRequestAction: Integer = BAD_ACTION_DISCARD;
  BadServerResponceAction: Integer = BAD_ACTION_DISCARD;
  LocalHostForClient: TSocketHost; // Ëîêàëüíûé àäðåñ, ñìîòðÿùèé íà êëèåíòà
  LocalHostForServer: TSocketHost; // Ëîêàëüíûé àäðåñ, ñìîòðÿùèé íà ñåðâåð

implementation

uses
  StrUtils, WinSock, Log;

var
  ProxyLocker: TRTLCriticalSection;

procedure LockProxy;
begin
  EnterCriticalSection(ProxyLocker);
end;

procedure UnlockProxy;
begin
  LeaveCriticalSection(ProxyLocker);
end;

{$region 'TProxyClient'}
constructor TProxyClient.Create(AClient: TCustomIpClient);
begin
  FClient := AClient;

  // Êëàññ ñîêåòà íà ñàìîì äåëå íå óìååò îïðåäåëÿòü àäðåñ ñåòåâîãî èíòåðôåéñà,
  // íà êîòîðûé çàáèíäåí ñîêåò. Ôóíêöèÿ LocalHostAddr õîòü è ÿâëÿåòñÿ ìåòîäîì,
  // íà ñàìîì äåëå íèêàêîé èíôîðìàöèè î ñîêåòå, êàæåòñÿ, íå èñïîëüçóåò.
  //  êîíå÷íîì èòîãå âñå ñâîäèòñÿ ê âûçîâó WSA ôóíêöèè gethostname().
  // Ïîýòîìó èñïîëüçóåì ëîêàëüíûé àäðåñ, âçÿòûé èç êîíôèãà - LocalHostForClient.
  FLocalHost := LocalHostForClient;
  FLocalPort := FClient.LocalPort;
end;

destructor TProxyClient.Destroy;
begin
  FServer.Free;
  FServerThread.Free;
  FProxyTube.Free;
  inherited;
end;

procedure TProxyClient.Stop;
begin
  FServer.Close;
  // Âñå îñòàëüíîå çàêðûâàåòñÿ àâòîìàòè÷åñêè ïî öåïî÷êå
end;

procedure TProxyClient.ConnectToServer(const Resource: String);
var
  Res: TRtspResource;
begin
  Resources.Lock;
  try
    Res := Resources.GetResource(Resource);
    if not Assigned(Res) then
      raise Exception.Create('Unknown resource: ' + Resource);

    FServerHost := Res.Host;
    FServerPort := Res.Port;
    FServerResource := Res.Name;
    FLocalResource := Resource;
  finally
    Resources.Unlock;
  end;

  // Ñîçäàåì ñîêåò, ñìîòðÿùèé íà ñåðâåð
  //
  FServer := TTcpClient.Create(nil);
  FServer.LocalHost := LocalHostForServer; // íîìåð ïîðòà ïóñòü âûáèðàåòñÿ àâòîìàòè÷åñêè
  FServer.RemoteHost := FServerHost;
  FServer.RemotePort := FServerPort;
  FServer.Open;

  if not FServer.Connected then
    raise Exception.CreateFmt('Could not connect to server %s:%s. WSA error: %d',
      [FServerHost, FServerPort, WSAGetLastError]);

  FProxyTube := TRtspProxyTube.Create(
    FClient.RemoteHost, LocalHostForClient,
    FServer.RemoteHost, LocalHostForServer);

  FServerThread := TServerThread.Create(True);
  FServerThread.FOwner := Self;
  FServerThread.FClient := FClient;
  FServerThread.FServer := FServer;
  FServerThread.Resume;
end;

procedure TProxyClient.Listen;
var
  Buf: array[0..4095] of Char;
  Count: Integer;
  ClientStr, Request: String;
  Msg: TRtspMessage;
  Setup1Done: Boolean;
  Transport: TRtspTransportHeader;
begin
  ClientStr := FClient.RemoteHost + ':' + FClient.RemotePort;
  WriteLine('[%s] Client session opened', [ClientStr]);

  Setup1Done := False;
  while True do
  try
    Count := FClient.ReceiveBuf(buf, 4096);
    if Count <= 0 then
      raise ESocketError.Create;

    SetString(Request, buf, count);
    WriteLine('[%s] Client request (%d bytes):%s%s',
      [ClientStr, Count, CRLF, Request], ClientColor);

    try
      Msg := TRtspMessage.Create(Request);
    except
      on e: Exception do
        case BadClientRequestAction of
          BAD_ACTION_DISCARD:
          begin
            WriteError('[%s] %s DISCARDED', [ClientStr, e.Message]);
            Continue;
          end;

          BAD_ACTION_IGNORE:
          begin
            WriteError('[%s] %s IGNORED', [ClientStr, e.Message]);
            FServer.SendLn(Request, '');
            Continue;
          end

          else raise;
        end;
    end;

    if not Assigned(FServer) then
    begin
      ConnectToServer(Msg.Resource);
      WriteLine('[%s] Server connected', [ClientStr]);
    end;

    if Msg.Method = 'SETUP' then
    begin
      Transport := Msg.GetTransportHeader;
      if not Assigned(Transport) then
        raise Exception.Create('Field "Transport" is not specified in client RTSP-request');

      if not Setup1Done then
      begin
        Setup1Done := True;

        // Ïîðòû, êîòîðûå ïðèñëàë êëèåíò, áóäóò óäàëåííûìè ïîðòàìè êëèåíòñêîé ñòîðîíû "òðóáû"
        FProxyTube.ProxyData1.RemotePort := Transport.ClientPortData;
        FProxyTube.ProxySync1.RemotePort := Transport.ClientPortSync;

        // À ê ñåðâåðó êîííåêòèìñÿ ëîêàëüíûìè ïîðòàìè ñåðâåðíîé ñòîðîíû "òðóáû"
        Transport.ClientPortData := FProxyTube.ServerData1.LocalPort;
        Transport.ClientPortSync := FProxyTube.ServerSync1.LocalPort;

        FServerThread.FOnResponce := ProcessSetup1Responce;
      end
      else
      begin
        // Ïîðòû, êîòîðûå ïðèñëàë êëèåíò, áóäóò óäàëåííûìè ïîðòàìè êëèåíòñêîé ñòîðîíû "òðóáû"
        FProxyTube.ProxyData2.RemotePort := Transport.ClientPortData;
        FProxyTube.ProxySync2.RemotePort := Transport.ClientPortSync;

        // À ê ñåðâåðó êîííåêòèìñÿ ëîêàëüíûìè ïîðòàìè ñåðâåðíîé ñòîðîíû "òðóáû"
        Transport.ClientPortData := FProxyTube.ServerData2.LocalPort;
        Transport.ClientPortSync := FProxyTube.ServerSync2.LocalPort;

        FServerThread.FOnResponce := ProcessSetup2Responce;
      end;
    end
    else if Msg.Method = 'PLAY' then
    begin
      FServerThread.FOnResponce := ProcessPlayResponce;
    end
    else if Msg.Method = 'TEARDOWN' then
    begin
      if Assigned(FProxyTube) then FProxyTube.Stop;
    end;

    // Ïîäìåíÿåì àäðåñ ïðîêñè (ò.å. ëîêàëüíûé àäðåñ è ïîðò) íà àäðåñ ðåàëüíîãî ñåðâåðà
    Msg.Address := FServerHost;
    Msg.Port := FServerPort;
    // Çàìåíà, à íå ïðîñòîå ïðèñâîåíèå, íóæíà ò.ê. âî âðåìÿ SETUP
    // èìÿ ðåñóðñà ðàñøèðåíî. Íàïðèìåð: rtsp://192.168.2.25/Channel2/track1
    Msg.Resource := AnsiReplaceText(Msg.Resource, FLocalResource, FServerResource);
    // Ñîãëàñíî ñòàíäàðòó, ññûëêà íà ðåñóðñ âñåãäà äàåòñÿ â àáñîëþòíîé ôîðìå
    Msg.ReplaceAll(
      Format('rtsp://%s:%s/', [FLocalHost, FLocalPort]),
      Format('rtsp://%s:%s/', [FServerHost, FServerPort]));
    // Èíîãäà àäðåñ ìîæåò íå ñîäåðæàòü ïîðò, åñëè îí äåôîëòíûé. Íàïðèìåð,
    // VLC âîçâðàùàåò "Content-Base=rtsp://192.168.197.128:554/Channel1"
    // à ÀÂÊ (Live555) "Content-Base=rtsp://192.168.5.4/Channel1"
    Msg.ReplaceAll(
        Format('rtsp://{0}/', [FLocalHost]),
        Format('rtsp://{0}/', [FServerHost]));
    Msg.ReplaceAll(FLocalResource, FServerResource);

    Request := Msg.HeaderToString;

    WriteLine('[%s] Transformed client request (%d bytes):%s%s',
      [ClientStr, Length(Request), CRLF, Request], ClientProxyColor);

    FServer.SendLn(Request, ''); // TODO: check state(?)

    Msg.Free;
  except
    on e: ESocketError do
    begin
      case e.ErrorCode of
        WSAECONNRESET:
          WriteLine('[%s] Client is offline', [ClientStr]);
        WSAEINTR, WSAENOTSOCK, 0:
          WriteLine('[%s] Canceled', [ClientStr]);
        else
          WriteError('[%s][Receive] WSA error: %d', [ClientStr, e.ErrorCode]);
      end;
      Break;
    end;

    on e: Exception do
    begin
      if e.Message <> '' then
        WriteError('[%s][Receive] %s', [ClientStr, e.Message]);
      // Ïî óìîë÷àíèþ îøèáêà çàêàí÷èâàåò öèêë îáùåíèÿ ñ ñåðâåðîì.
      // Åñëè îøèáêà íå êðèòè÷íàÿ è îáùåíèå ìîæíî ïðîäîëæèòü,
      // òî íóæíî îáðàáîòàòü åå îòäåëüíî.
      Break;
    end;
  end;

  WriteLine('[%s] Client session closed', [ClientStr]);

  if Assigned(FProxyTube) then
    FProxyTube.Stop;

  FClient.Close;

  if Assigned(FServer) then
    FServer.Close;
  if Assigned(FServerThread) then
    FServerThread.WaitFor;
end;

procedure TServerThread.Execute;
var
  Buf: array[0..4095] of Char;
  Count: Integer;
  ServerStr, Responce: String;
  Msg: TRtspMessage;
begin
  ServerStr := 'Server -> ' + FClient.RemoteHost + ':' + FClient.RemotePort;
  WriteLine('[%s] Server session opened', [ServerStr]);

  while True do
  try
    Count := FServer.ReceiveBuf(Buf, 4096);
    if Count <= 0 then
      raise ESocketError.Create;

    SetString(Responce, Buf, Count);
    WriteLine('[%s] Server responce (%d bytes):%s%s',
      [ServerStr, Count, CRLF, Responce], ServerColor);

    try
      Msg := TRtspMessage.Create(Responce);
    except
      on e: Exception do
        case BadServerResponceAction of
          BAD_ACTION_DISCARD:
          begin
            WriteError('[%s] %s DISCARDED', [ServerStr, e.Message]);
            Continue;
          end;

          BAD_ACTION_IGNORE:
          begin
            WriteError('[%s] %s IGNORED', [ServerStr, e.Message]);
            FClient.SendLn(Responce, '');
            Continue;
          end

          else raise;
        end;
    end;

    // VLC ñíà÷àëà ïðèñûëàåò çàãîëîâîê, ïîòîì òåëî ñîîáùåíèÿ îòäåëüíûì ïàêåòîì
    while Msg.ContentLength > Length(Msg.Content) do
    begin
      WriteLine('[%s] Bytes received: %d; expecting more: %d',
        [ServerStr, Count, Msg.ContentLength - Length(Msg.Content)], ServerColor);

      Count := FServer.ReceiveBuf(Buf, 4096);
      if Count <= 0 then
        raise ESocketError.Create;

      SetString(Responce, Buf, Count);
      WriteLine('[%s] Server responce (%d bytes):%s%s',
        [ServerStr, Count, CRLF, Responce], ServerColor);

      Msg.Content := Msg.Content + Responce;
    end;

    if Assigned(FOnResponce) then
    begin
      FOnResponce(Self, Msg);
      FOnResponce := nil;
    end;

    // Ïîäìåíÿåì àäðåñ ñåðâåðà íà àäðåñ ïðîêñè (ò.å. ëîêàëüíûé àäðåñ è ïîðò)
    // Ñîãëàñíî ñòàíäàðòó, ññûëêà íà ðåñóðñ âñåãäà äàåòñÿ â àáñîëþòíîé ôîðìå
    Msg.ReplaceAll(
      Format('rtsp://%s:%s/', [FOwner.FServerHost, FOwner.FServerPort]),
      Format('rtsp://%s:%s/', [FOwner.FLocalHost, FOwner.FLocalPort]));
    // Èíîãäà àäðåñ ìîæåò íå ñîäåðæàòü ïîðò, åñëè îí äåôîëòíûé. Íàïðèìåð,
    // VLC âîçâðàùàåò "Content-Base=rtsp://192.168.197.128:554/Channel1"
    // à ÀÂÊ "Content-Base=rtsp://192.168.5.4/Channel1"
    Msg.ReplaceAll(
      Format('rtsp://%s/', [FOwner.FServerHost]),
      Format('rtsp://%s/', [FOwner.FLocalHost]));
    Msg.ReplaceAll(FOwner.FServerResource, FOwner.FLocalResource);

    Responce := Msg.HeaderToString;

    WriteLine('[%s] Transformed server responce (%d bytes):%s%s',
      [ServerStr, Length(Responce), CRLF, Responce], ServerProxyColor);

    FClient.SendLn(Responce, '');

    if Msg.Content <> '' then
    begin
      Responce := Msg.Content;

      WriteLine('[%s] Transformed server responce (%d bytes):%s%s',
        [ServerStr, Length(Responce), CRLF, Responce], ServerProxyColor);

      FClient.SendLn(Responce, ''); // TODO: check state(?)
    end;

    Msg.Free;
  except
    on e: ESocketError do
    begin
      case e.ErrorCode of
        WSAECONNRESET:
          WriteLine('[%s] Server is offline', [ServerStr]);
        WSAEINTR, WSAENOTSOCK, 0:
          WriteLine('[%s] Canceled', [ServerStr]);
        else
          WriteError('[%s][Receive] WSA error: %d', [ServerStr, e.ErrorCode]);
      end;
      Break;
    end;

    on e: Exception do
    begin
      if e.Message <> '' then
        WriteError('[%s][Receive]: %s', [ServerStr, e.Message]);
      // Ïî óìîë÷àíèþ îøèáêà çàêàí÷èâàåò öèêë îáùåíèÿ ñ ñåðâåðîì.
      // Åñëè îøèáêà íå êðèòè÷íàÿ è îáùåíèå ìîæíî ïðîäîëæèòü,
      // òî íóæíî îáðàáîòàòü åå îòäåëüíî.
      Break;
    end;
  end;

  WriteLine('[%s] Server session closed', [ServerStr]);

  if Assigned(FOwner.FProxyTube) then
    FOwner.FProxyTube.Stop;

  FServer.Close;
  FClient.Close;
end;

procedure TProxyClient.ProcessSetup1Responce(Sender: TObject; Msg: TRtspMessage);
var
  Transport: TRtspTransportHeader;
begin
	Transport := Msg.GetTransportHeader;
	if not Assigned(Transport) then
    raise Exception.Create('Server did not specify field "Transport" in responce to the SETUP request');

  FProxyTube.ServerData1.RemotePort := Transport.ServerPortData;
  FProxyTube.ServerSync1.RemotePort := Transport.ServerPortSync;

  Transport.ServerPortData := FProxyTube.ProxyData1.LocalPort;
  Transport.ServerPortSync := FProxyTube.ProxySync1.LocalPort;
  Transport.ClientPortData := FProxyTube.ProxyData1.RemotePort;
  Transport.ClientPortSync := FProxyTube.ProxySync1.RemotePort;
end;

procedure TProxyClient.ProcessSetup2Responce(Sender: TObject; Msg: TRtspMessage);
var
  Transport: TRtspTransportHeader;
begin
	Transport := Msg.GetTransportHeader;
	if not Assigned(Transport) then
    raise Exception.Create('Server did not specify field "Transport" in responce to the SETUP request');

  FProxyTube.ServerData2.RemotePort := Transport.ServerPortData;
  FProxyTube.ServerSync2.RemotePort := Transport.ServerPortSync;

  Transport.ServerPortData := FProxyTube.ProxyData2.LocalPort;
  Transport.ServerPortSync := FProxyTube.ProxySync2.LocalPort;
  Transport.ClientPortData := FProxyTube.ProxyData2.RemotePort;
  Transport.ClientPortSync := FProxyTube.ProxySync2.RemotePort;
end;

procedure TProxyClient.ProcessPlayResponce(Sender: TObject; Msg: TRtspMessage);
begin
  FProxyTube.Start;
end;
{$endregion}

{$region 'TProxyServer'}
constructor TProxyServer.Create(LocalPort: TSocketPort);
begin
  InitializeCriticalSection(FClientsLocker);

  FClients := TList.Create;

  FProxy := TTcpServer.Create(nil);
  FProxy.LocalHost := LocalHostForClient;
  FProxy.LocalPort := LocalPort;
  FProxy.OnAccept := ClientAccepted;
end;

destructor TProxyServer.Destroy;
begin
  Stop;

  FClients.Free;
  FProxy.Free;

  DeleteCriticalSection(FClientsLocker);
  inherited;
end;

procedure TProxyServer.Start;
begin
  if not FProxy.Active then
  begin
    FProxy.Open;

    WriteLine('Proxy started at %s:%s', [FProxy.LocalHost, FProxy.LocalPort]);
  end;
end;

procedure TProxyServer.Stop;
begin
  if FProxy.Active then
  begin
    while FClients.Count > 0 do
      TProxyClient(FClients[0]).Stop;
    FProxy.Close;

    WriteLine('Proxy stopped at %s:%s', [FProxy.LocalHost, FProxy.LocalPort]);
  end;
end;

// Ìåòîä âûïîëíÿåòñÿ â îòäåëüíîì ïîòîêå, àâòîìàòè÷åñêè ñîçäàâàåìîì TCP-ñåðâåðîì.
// Ïîñëå çàâåðøåíèÿ ìåòîäà ñåðâåð çàêðîåò êëèåíòñêîé ñîêåò, óíè÷òîæèò îáúåêò AClient,
// à ïîòîê çàñíåò è áóäåò ïîâòîðíî èñïîëüçîâàí ïðè ïîäêëþ÷åíèè äðóãîãî êëèåíòà.
// TODO: ×èñëî ïîòîêîâ â ïóëå TCP-ñåðâåðà îãðàíè÷åíî (10), íóæíî êàê-òî ðàçðóëèòü
// ñèòóàöèþ, åñëè âäðóã îäíîâðåìåííî çàõîòÿò ïîäêëþ÷èòüñÿ áîëüøå êëèåíòîâ.
procedure TProxyServer.ClientAccepted(Sender: TObject; Client: TCustomIpClient);
var
  ClientStr: String;
  ProxyClient: TProxyClient;
begin
  ClientStr := Client.RemoteHost + ':' + Client.RemotePort;
  try
    WriteLine('Client Accepted: ' + ClientStr);

    ProxyClient := TProxyClient.Create(Client);
    try
      AddClient(ProxyClient);
      ProxyClient.Listen;
      RemoveClient(ProxyClient);
    finally
      ProxyClient.Free;
    end;
  except
    on e: Exception do
      WriteError('[%s][Accept] %s', [ClientStr, e.Message]);
  end;
end;

procedure TProxyServer.AddClient(AClient: TProxyClient);
begin
  EnterCriticalSection(FClientsLocker);
  try
    FClients.Add(AClient);
  finally
    LeaveCriticalSection(FClientsLocker);
  end;
end;

procedure TProxyServer.RemoveClient(AClient: TProxyClient);
begin
  EnterCriticalSection(FClientsLocker);
  try
    FClients.Remove(AClient);
  finally
    LeaveCriticalSection(FClientsLocker);
  end;
end;
{$endregion}

initialization
  InitializeCriticalSection(ProxyLocker);

finalization
  DeleteCriticalSection(ProxyLocker);

end.