unit AsyncReader;

   (*********************************************************************
    * The contents of this file are used with permission, subject to    *
    * the Mozilla Public License Version 1.1 (the "License"); you may   *
    * not use this file except in compliance with the License. You may  *
    * obtain a copy of the License at                                   *
    * http://www.mozilla.org/MPL/MPL-1.1.html                           *
    *                                                                   *
    * Software distributed under the License is distributed on an       *
    * "AS IS" basis, WITHOUT WARRANTY OF ANY KIND, either express or    *
    * implied. See the License for the specific language governing      *
    * rights and limitations under the License.                         *
    *                                                                   *
    * (C) 2004 Martin Offenwanger: coder@dsplayer.de                    *
    *********************************************************************)
{
@author(Martin Offenwanger: coder@dsplayer.de)
@created(Apr 22, 2004)
@lastmod(Sep 09, 2004)
}

interface

uses
  ActiveX, Classes, DirectShow9, BaseClass, Windows, Queue, Config, Forms,
  ShoutCastStream, SysUtils, Dialogs, ExtCtrls;

type
  TAsyncIO = class(TInterfacedObject, IAsyncReader)
  private
    FStream: IStream;
    FStop,
      FWaiting,
      FFlushing,
      FFwdStream: boolean;
    FReaderLock,
      FListsLock: TBCCritSec;
    FWorkList,
      FDoneList: TQueue;
    FWorkEvent,
      FDoneEvent,
      FAllDoneEv: TBCAMEvent;
    FOutCount: Longint;
    FStrmSize: Int64;
    FThread: TThread;
    FURLMode: boolean;
    FMediaControl: IMediaControl;
    { the pause and run commands called with FMediaControl in Syncread
      must called via a timer, otherwise ondestroy in unit filter won't called }
    FTimerPlay: TTimer;
    FTimerPause: TTimer;
    procedure OnTimerPlay(Sender: TObject);
    procedure OnTimerPause(Sender: TObject);
    procedure PutDoneItem(AItem: PAsyncRequest);
    function GetDoneItem: PAsyncRequest;
    function PutWorkItem(AItem: PAsyncRequest): HRESULT;
    function GetWorkItem: PAsyncRequest;
    function SetPosition(const APos: Int64): HResult;
    procedure InitStreamLen;
    function SetStreamPos(const APos: Int64): HResult;
    function GetStreamPos: Int64;
    function CreateRequest(llPos: LONGLONG; lLength: Integer;
      bAligned: BOOL; pBuffer: Pointer; pContext: Pointer;
      dwUser: DWORD): PAsyncRequest;
    procedure CompleteRequest(Req: PAsyncRequest);
    function InitAllocator(out Alloc: IMemAllocator): HRESULT; virtual;
    function DoRequest(llPos: LONGLONG; lLength: Longint;
      bAligned: BOOL; pBuffer: Pointer; pContext: Pointer;
      dwUser: DWORD): HResult;
    function DoWaitForNext(dwTimeout: DWORD; var ppContext: Pointer;
      var pdwUser: DWORD; var pcbActual: Longint): HRESULT;
  protected
    // IAsyncReader methods
    function RequestAllocator(pPreferred: IMemAllocator;
      pProps: PAllocatorProperties;
      out ppActual: IMemAllocator): HResult; stdcall;
    function Request(pSample: IMediaSample; dwUser: DWORD): HResult; stdcall;
    function WaitForNext(dwTimeout: DWORD; out ppSample: IMediaSample;
      out pdwUser: DWORD): HResult; stdcall;
    function SyncReadAligned(pSample: IMediaSample): HResult; stdcall;
    function SyncRead(llPosition: int64; lLength: Longint;
      pBuffer: Pbyte): HResult; stdcall;
    function Length(out pTotal, pAvailable: int64): HResult; stdcall;
  public
    constructor Create(AStream: IStream; FwdOnly: boolean = false;
      const StreamSize: Int64 = 0; URLMode: boolean = false);
    // calling the destructor causes crashes
    destructor Destroy; override;
    // we use this function to detroy memeber objects
    procedure FreeAllObjects;
    // the graph object for full control during buffering URL stream
    procedure SetActiveGraph(var FilterGraph: IFilterGraph);
    procedure Addref;
    procedure Release;
    procedure Process;
    // IAsyncReader methods
    function BeginFlush: HRESULT; stdcall;
    function EndFlush: HRESULT; stdcall;
    // FURLMode methods
    procedure Connect(Adress: string; Port: string;
      Location: string; MetaData: boolean);
  end;

implementation

uses WorkerThread, filter;

procedure TAsyncIO.setActiveGraph(var FilterGraph: IFilterGraph);
begin
  // In URlmode we need to control the Graph during buffering
  if (FURLMode) and (FMediaControl = nil) then
  begin
    FilterGraph.QueryInterface(IID_IMediaControl, FMediaControl);
    FTimerPlay := TTimer.Create(nil);
    FTimerPlay.Enabled := false;
    FTimerPlay.Interval := 1;
    // makes shure that run is always called after pause
    FTimerPlay.OnTimer := OnTimerPlay;
    FTimerPause := TTimer.Create(nil);
    FTimerPause.Enabled := false;
    FTimerPause.Interval := 1;
    FTimerPause.OnTimer := OnTimerPause;
  end;
end;

procedure TAsyncIO.Connect(Adress: string; Port: string; Location: string;
  MetaData: boolean);
begin
  GFExit := false;
  g_threadedShoutCastStream := TThreadedShoutcastStream.Create(Adress, Port,
    Location, MetaData);
end;

procedure TAsyncIO.Release;
begin
  _Release;
end;

procedure TAsyncIO.Addref;
begin
  _AddRef;
end;

constructor TAsyncIO.Create(AStream: IStream; FwdOnly: boolean = false;
  const StreamSize: Int64 = 0; URLMode: boolean = false);
begin
  inherited Create;
  FTimerPlay := nil;
  if g_threadedShoutCastStream <> nil then
  begin
    g_threadedShoutCastStream.Destroy;
    g_threadedShoutCastStream := nil;
  end;
  FURLMode := URLMode;
  FStream := AStream;
  FListsLock := TBCCritSec.Create;
  FReaderLock := TBCCritSec.Create;
  FWorkList := TQueue.Create;
  FDoneList := TQueue.Create;
  FWorkEvent := TBCAMEvent.Create(true);
  FDoneEvent := TBCAMEvent.Create(true);
  FAllDoneEv := TBCAMEvent.Create(true);
  FFwdStream := FwdOnly;
  FStrmSize := StreamSize;
  FWorkEvent.Reset;
  FThread := TWorkThread.Create(Self);
  FThread.Resume;
end;

procedure TAsyncIO.FreeAllObjects;
var
  Req: PAsyncRequest;
begin
  FStop := true;
  FThread.Terminate;
  FWorkEvent.SetEv;
  FThread.WaitFor;
  FThread.Free;
  Req := GetDoneItem;
  while Req <> nil do
  begin
    Dispose(Req);
    Req := GetDoneItem;
  end;
  // FStream._Release;
  FReaderLock.Free;
  FListsLock.Free;
  FWorkList.Free;
  FDoneList.Free;
  FWorkEvent.Free;
  FDoneEvent.Free;
  FAllDoneEv.Free;
  FTimerPlay.Free;
  FTimerPause.Free;
end;

destructor TAsyncIO.Destroy;
var
  Req: PAsyncRequest;
begin
  GFExit := true;
  FStop := true;
  FThread.Terminate;
  FWorkEvent.SetEv;
  FThread.WaitFor;
  FThread.Free;
  Req := GetDoneItem;
  while Req <> nil do
  begin
    Dispose(Req);
    Req := GetDoneItem;
  end;
  FStream := nil;
  FReaderLock.Free;
  FListsLock.Free;
  FWorkList.Free;
  FDoneList.Free;
  FWorkEvent.Free;
  FDoneEvent.Free;
  FAllDoneEv.Free;
  inherited destroy;
end;

function TAsyncIO.BeginFlush: HRESULT;
var
  Req: PAsyncRequest;
begin
  GFExit := true;
  { need to nil here IMediaControl,
    if not, the destructor in TFilter will not executed }
  FMediaControl := nil;
  FListsLock.Lock;
  Result := S_OK;
  // we nil here and in the filter destructor
  if g_threadedShoutCastStream <> nil then
  begin
    g_threadedShoutCastStream.Destroy;
    g_threadedShoutCastStream := nil;
  end;
  if GFStringQueue <> nil then
  begin
    GFStringQueue.destroy;
    GFStringQueue := nil;
  end;
  try
    FFlushing := true;
    Req := GetWorkItem;
    while Req <> nil do
    begin
      PutDoneItem(Req);
      Req := GetWorkItem;
    end;
    if FOutCount > 0 then
    begin
      Assert(not FWaiting);
      FAllDoneEv.Reset;
      FWaiting := true;
    end
    else
    begin
      FDoneEvent.SetEv;
      FWorkEvent.SetEv;
    end;
  finally
    FListsLock.UnLock;
  end;
  //Assert(FWaiting);
  while FWaiting do
  begin
    FAllDoneEv.Wait();
    FListsLock.Lock;
    try
      if FOutCount = 0 then
      begin
        FWaiting := false;
        FDoneEvent.SetEv;
      end;
    finally
      FListsLock.UnLock;
    end;
  end;
end;

function TAsyncIO.EndFlush: HRESULT;
begin
  GFExit := true;
  FListsLock.Lock;
  FFlushing := false;
  Assert(not FWaiting);

  if FDoneList.Count > 0 then
    FDoneEvent.SetEv
  else
    FDoneEvent.Reset;

  Result := S_OK;
  FListsLock.UnLock;
end;

procedure TAsyncIO.Process;
var
  Req: PAsyncRequest;
begin
  while true do
  begin
    FWorkEvent.Wait;
    FListsLock.Lock;
    Req := GetWorkItem;
    if Req <> nil then
      Inc(FOutCount);
    FListsLock.UnLock;

    if Req <> nil then
    begin
      CompleteRequest(Req);
      FListsLock.Lock;
      PutDoneItem(Req);
      Dec(FOutCount);
      if (FOutCount = 0) and FWaiting then
        FAllDoneEv.SetEv;
      FListsLock.UnLock;
    end;
    if FStop then
      break;
  end;
end;

function TAsyncIO.DoRequest(
  llPos: LONGLONG; lLength: Integer; bAligned: BOOL; pBuffer,
  pContext: Pointer; dwUser: DWORD): HResult;
var
  Req: PAsyncRequest;
begin
  Req := CreateRequest(llPos, lLength, bAligned, pBuffer, pContext, dwUser);
  Result := PutWorkItem(Req);
  if not Succeeded(Result) then
    Dispose(Req);
end;

function TAsyncIO.DoWaitForNext(dwTimeout: DWORD; var ppContext: Pointer;
  var pdwUser: DWORD; var pcbActual: Integer): HRESULT;
var
  Req: PAsyncRequest;
begin
  Result := S_OK;
  ppContext := nil;
  pdwUser := 0;
  pcbActual := 0;
  while true do
  begin
    if (not FDoneEvent.Wait(dwTimeout)) then
    begin
      Result := VFW_E_TIMEOUT;
      Break;
    end;
    Req := GetDoneItem;
    if Req <> nil then
    begin
      ppContext := Req.FContext;
      pdwUser := Req.FUser;
      pcbActual := Req.FLength;
      Result := Req.Fhr;
      Dispose(Req);
      Break;
    end
    else
    begin
      FListsLock.Lock;
      try
        if FFlushing {and not FWaiting} then
        begin
          Result := VFW_E_WRONG_STATE;
          Break;
        end;
      finally
        FListsLock.UnLock;
      end;
    end;
  end;
end;

procedure TAsyncIO.OnTimerPlay(Sender: TObject);
begin
  if FMediaControl <> nil then
    FMediaControl.Run;
  FTimerPlay.Enabled := false;
end;

procedure TAsyncIO.OnTimerPause(Sender: TObject);
begin
  if FMediaControl <> nil then
    FMediaControl.Pause;
  FTimerPause.Enabled := false;
end;

function TAsyncIO.SyncRead(llPosition: int64; lLength: Longint;
  pBuffer: Pbyte): HResult;
var
  Req: PAsyncRequest;
  DataWritten: boolean;
  i: integer;
  StringStream: TStringStream;
  Buffer: string;
  Tempbuffer: string;
  Avdata: int64;
  Application: TApplication;
  Buffering: boolean;
  Count: integer;
begin
  // we do not accept a Nil buffer
  if pBuffer = nil then
  begin
    result := E_FAIL;
    exit;
  end;
  Result := S_OK;
  // the URL buffer control for Dirctshow is added here
  // buffering during the playback
  if FURLMode then
  begin
    // the min. buffersize must be equal to the requested length
    if GFBufferSize < lLength then
      GFBufferSize := lLength;
    // Mpeg1 splitter requests same samples during connection process and
    // after starting the graph.
    StringStream := nil;
    GFStreamPos := llPosition;
    DataWritten := false;
    Buffer := '';
    Tempbuffer := '';
    Avdata := 0;
    Buffering := false;
    Count := 0;
    Application := TApplication.Create(nil);
    if not GFConnected then
    begin
      if assigned(GFFilterCallBack) then
        GFFilterCallBack.AsyncExFilterState(false, false, true, false, 0);

      // since XP ServicePack2 rc2 the mpeg splitter requests a end sample
      // of the stream during pin connection process,
      // we skip this sample because we can't send it
      if (llPosition > (GCFInt64max - lLength - 2)) then
      begin
        result := E_FAIL;
        exit;
      end;
      i := 0;
      if GFStringQueue = nil then
      begin
        result := E_FAIL;
        exit;
      end;
      while not Datawritten do
      begin
        if GFStringQueue <> nil then
          Count := GFStringQueue.getcount;
        if ((GFExit) or (GFStringQueue = nil) or (Count <= i)) then