reactos/dll/directx/wine/quartz/pin.c

   1 /*
   2  * Generic Implementation of IPin Interface
   3  *
   4  * Copyright 2003 Robert Shearman
   5  *
   6  * This library is free software; you can redistribute it and/or
   7  * modify it under the terms of the GNU Lesser General Public
   8  * License as published by the Free Software Foundation; either
   9  * version 2.1 of the License, or (at your option) any later version.
  10  *
  11  * This library is distributed in the hope that it will be useful,
  12  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  14  * Lesser General Public License for more details.
  15  *
  16  * You should have received a copy of the GNU Lesser General Public
  17  * License along with this library; if not, write to the Free Software
  18  * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA
  19  */
  20
  21 #include "quartz_private.h"
  22
  23 static const IPinVtbl PullPin_Vtbl;
  24
  25 #define ALIGNDOWN(value,boundary) ((value)/(boundary)*(boundary))
  26 #define ALIGNUP(value,boundary) (ALIGNDOWN((value)+(boundary)-1, (boundary)))
  27
  28 typedef HRESULT (*SendPinFunc)( IPin *to, LPVOID arg );
  29
  30 /** Helper function, there are a lot of places where the error code is inherited
  31  * The following rules apply:
  32  *
  33  * Return the first received error code (E_NOTIMPL is ignored)
  34  * If no errors occur: return the first received non-error-code that isn't S_OK
  35  */
  36 static HRESULT updatehres( HRESULT original, HRESULT new )
  37 {
  38     if (FAILED( original ) || new == E_NOTIMPL)
  39         return original;
  40
  41     if (FAILED( new ) || original == S_OK)
  42         return new;
  43
  44     return original;
  45 }
  46
  47 /** Sends a message from a pin further to other, similar pins
  48  * fnMiddle is called on each pin found further on the stream.
  49  * fnEnd (can be NULL) is called when the message can't be sent any further (this is a renderer or source)
  50  *
  51  * If the pin given is an input pin, the message will be sent downstream to other input pins
  52  * If the pin given is an output pin, the message will be sent upstream to other output pins
  53  */
  54 static HRESULT SendFurther( IPin *from, SendPinFunc fnMiddle, LPVOID arg, SendPinFunc fnEnd )
  55 {
  56     PIN_INFO pin_info;
  57     ULONG amount = 0;
  58     HRESULT hr = S_OK;
  59     HRESULT hr_return = S_OK;
  60     IEnumPins *enumpins = NULL;
  61     BOOL foundend = TRUE;
  62     PIN_DIRECTION from_dir;
  63
  64     IPin_QueryDirection( from, &from_dir );
  65
  66     hr = IPin_QueryInternalConnections( from, NULL, &amount );
  67     if (hr != E_NOTIMPL && amount)
  68         FIXME("Use QueryInternalConnections!\n");
  69
  70     pin_info.pFilter = NULL;
  71     hr = IPin_QueryPinInfo( from, &pin_info );
  72     if (FAILED(hr))
  73         goto out;
  74
  75     hr = IBaseFilter_EnumPins( pin_info.pFilter, &enumpins );
  76     if (FAILED(hr))
  77         goto out;
  78
  79     hr = IEnumPins_Reset( enumpins );
  80     while (hr == S_OK) {
  81         IPin *pin = NULL;
  82         hr = IEnumPins_Next( enumpins, 1, &pin, NULL );
  83         if (hr == VFW_E_ENUM_OUT_OF_SYNC)
  84         {
  85             hr = IEnumPins_Reset( enumpins );
  86             continue;
  87         }
  88         if (pin)
  89         {
  90             PIN_DIRECTION dir;
  91
  92             IPin_QueryDirection( pin, &dir );
  93             if (dir != from_dir)
  94             {
  95                 IPin *connected = NULL;
  96
  97                 foundend = FALSE;
  98                 IPin_ConnectedTo( pin, &connected );
  99                 if (connected)
 100                 {
 101                     HRESULT hr_local;
 102
 103                     hr_local = fnMiddle( connected, arg );
 104                     hr_return = updatehres( hr_return, hr_local );
 105                     IPin_Release(connected);
 106                 }
 107             }
 108             IPin_Release( pin );
 109         }
 110         else
 111         {
 112             hr = S_OK;
 113             break;
 114         }
 115     }
 116
 117     if (!foundend)
 118         hr = hr_return;
 119     else if (fnEnd) {
 120         HRESULT hr_local;
 121
 122         hr_local = fnEnd( from, arg );
 123         hr_return = updatehres( hr_return, hr_local );
 124     }
 125
 126 out:
 127     if (enumpins)
 128         IEnumPins_Release( enumpins );
 129     if (pin_info.pFilter)
 130         IBaseFilter_Release( pin_info.pFilter );
 131     return hr;
 132 }
 133
 134
 135 static void Copy_PinInfo(PIN_INFO * pDest, const PIN_INFO * pSrc)
 136 {
 137     /* Tempting to just do a memcpy, but the name field is
 138        128 characters long! We will probably never exceed 10
 139        most of the time, so we are better off copying
 140        each field manually */
 141     strcpyW(pDest->achName, pSrc->achName);
 142     pDest->dir = pSrc->dir;
 143     pDest->pFilter = pSrc->pFilter;
 144 }
 145
 146 static HRESULT deliver_endofstream(IPin* pin, LPVOID unused)
 147 {
 148     return IPin_EndOfStream( pin );
 149 }
 150
 151 static HRESULT deliver_beginflush(IPin* pin, LPVOID unused)
 152 {
 153     return IPin_BeginFlush( pin );
 154 }
 155
 156 static HRESULT deliver_endflush(IPin* pin, LPVOID unused)
 157 {
 158     return IPin_EndFlush( pin );
 159 }
 160
 161 typedef struct newsegmentargs
 162 {
 163     REFERENCE_TIME tStart, tStop;
 164     double rate;
 165 } newsegmentargs;
 166
 167 static HRESULT deliver_newsegment(IPin *pin, LPVOID data)
 168 {
 169     newsegmentargs *args = data;
 170     return IPin_NewSegment(pin, args->tStart, args->tStop, args->rate);
 171 }
 172
 173 /*** PullPin implementation ***/
 174
 175 static HRESULT PullPin_Init(const IPinVtbl *PullPin_Vtbl, const PIN_INFO * pPinInfo, SAMPLEPROC_PULL pSampleProc, LPVOID pUserData,
 176                             QUERYACCEPTPROC pQueryAccept, CLEANUPPROC pCleanUp, REQUESTPROC pCustomRequest, STOPPROCESSPROC pDone, LPCRITICAL_SECTION pCritSec, PullPin * pPinImpl)
 177 {
 178     /* Common attributes */
 179     pPinImpl->pin.IPin_iface.lpVtbl = PullPin_Vtbl;
 180     pPinImpl->pin.refCount = 1;
 181     pPinImpl->pin.pConnectedTo = NULL;
 182     pPinImpl->pin.pCritSec = pCritSec;
 183     Copy_PinInfo(&pPinImpl->pin.pinInfo, pPinInfo);
 184     ZeroMemory(&pPinImpl->pin.mtCurrent, sizeof(AM_MEDIA_TYPE));
 185
 186     /* Input pin attributes */
 187     pPinImpl->pUserData = pUserData;
 188     pPinImpl->fnQueryAccept = pQueryAccept;
 189     pPinImpl->fnSampleProc = pSampleProc;
 190     pPinImpl->fnCleanProc = pCleanUp;
 191     pPinImpl->fnDone = pDone;
 192     pPinImpl->fnPreConnect = NULL;
 193     pPinImpl->pAlloc = NULL;
 194     pPinImpl->prefAlloc = NULL;
 195     pPinImpl->pReader = NULL;
 196     pPinImpl->hThread = NULL;
 197     pPinImpl->hEventStateChanged = CreateEventW(NULL, TRUE, TRUE, NULL);
 198     pPinImpl->thread_sleepy = CreateEventW(NULL, FALSE, FALSE, NULL);
 199
 200     pPinImpl->rtStart = 0;
 201     pPinImpl->rtCurrent = 0;
 202     pPinImpl->rtStop = ((LONGLONG)0x7fffffff << 32) | 0xffffffff;
 203     pPinImpl->dRate = 1.0;
 204     pPinImpl->state = Req_Die;
 205     pPinImpl->fnCustomRequest = pCustomRequest;
 206     pPinImpl->stop_playback = 1;
 207
 208     InitializeCriticalSection(&pPinImpl->thread_lock);
 209     pPinImpl->thread_lock.DebugInfo->Spare[0] = (DWORD_PTR)( __FILE__ ": PullPin.thread_lock");
 210
 211     return S_OK;
 212 }
 213
 214 HRESULT PullPin_Construct(const IPinVtbl *PullPin_Vtbl, const PIN_INFO * pPinInfo, SAMPLEPROC_PULL pSampleProc, LPVOID pUserData, QUERYACCEPTPROC pQueryAccept, CLEANUPPROC pCleanUp, REQUESTPROC pCustomRequest, STOPPROCESSPROC pDone, LPCRITICAL_SECTION pCritSec, IPin ** ppPin)
 215 {
 216     PullPin * pPinImpl;
 217
 218     *ppPin = NULL;
 219
 220     if (pPinInfo->dir != PINDIR_INPUT)
 221     {
 222         ERR("Pin direction(%x) != PINDIR_INPUT\n", pPinInfo->dir);
 223         return E_INVALIDARG;
 224     }
 225
 226     pPinImpl = CoTaskMemAlloc(sizeof(*pPinImpl));
 227
 228     if (!pPinImpl)
 229         return E_OUTOFMEMORY;
 230
 231     if (SUCCEEDED(PullPin_Init(PullPin_Vtbl, pPinInfo, pSampleProc, pUserData, pQueryAccept, pCleanUp, pCustomRequest, pDone, pCritSec, pPinImpl)))
 232     {
 233         *ppPin = &pPinImpl->pin.IPin_iface;
 234         return S_OK;
 235     }
 236
 237     CoTaskMemFree(pPinImpl);
 238     return E_FAIL;
 239 }
 240
 241 static HRESULT PullPin_InitProcessing(PullPin * This);
 242
 243 HRESULT WINAPI PullPin_ReceiveConnection(IPin * iface, IPin * pReceivePin, const AM_MEDIA_TYPE * pmt)
 244 {
 245     PIN_DIRECTION pindirReceive;
 246     HRESULT hr = S_OK;
 247     PullPin *This = impl_PullPin_from_IPin(iface);
 248
 249     TRACE("(%p/%p)->(%p, %p)\n", This, iface, pReceivePin, pmt);
 250     dump_AM_MEDIA_TYPE(pmt);
 251
 252     EnterCriticalSection(This->pin.pCritSec);
 253     if (!This->pin.pConnectedTo)
 254     {
 255         ALLOCATOR_PROPERTIES props;
 256
 257         props.cBuffers = 3;
 258         props.cbBuffer = 64 * 1024; /* 64 KB */
 259         props.cbAlign = 1;
 260         props.cbPrefix = 0;
 261
 262         if (This->fnQueryAccept(This->pUserData, pmt) != S_OK)
 263             hr = VFW_E_TYPE_NOT_ACCEPTED; /* FIXME: shouldn't we just map common errors onto
 264                                            * VFW_E_TYPE_NOT_ACCEPTED and pass the value on otherwise? */
 265
 266         if (SUCCEEDED(hr))
 267         {
 268             IPin_QueryDirection(pReceivePin, &pindirReceive);
 269
 270             if (pindirReceive != PINDIR_OUTPUT)
 271             {
 272                 ERR("Can't connect from non-output pin\n");
 273                 hr = VFW_E_INVALID_DIRECTION;
 274             }
 275         }
 276
 277         This->pReader = NULL;
 278         This->pAlloc = NULL;
 279         This->prefAlloc = NULL;
 280         if (SUCCEEDED(hr))
 281         {
 282             hr = IPin_QueryInterface(pReceivePin, &IID_IAsyncReader, (LPVOID *)&This->pReader);
 283         }
 284
 285         if (SUCCEEDED(hr) && This->fnPreConnect)
 286         {
 287             hr = This->fnPreConnect(iface, pReceivePin, &props);
 288         }
 289
 290         /*
 291          * Some custom filters (such as the one used by Fallout 3
 292          * and Fallout: New Vegas) expect to be passed a non-NULL
 293          * preferred allocator.
 294          */
 295         if (SUCCEEDED(hr))
 296         {
 297             hr = StdMemAllocator_create(NULL, (LPVOID *) &This->prefAlloc);
 298         }
 299
 300         if (SUCCEEDED(hr))
 301         {
 302             hr = IAsyncReader_RequestAllocator(This->pReader, This->prefAlloc, &props, &This->pAlloc);
 303         }
 304
 305         if (SUCCEEDED(hr))
 306         {
 307             CopyMediaType(&This->pin.mtCurrent, pmt);
 308             This->pin.pConnectedTo = pReceivePin;
 309             IPin_AddRef(pReceivePin);
 310             hr = IMemAllocator_Commit(This->pAlloc);
 311         }
 312
 313         if (SUCCEEDED(hr))
 314             hr = PullPin_InitProcessing(This);
 315
 316         if (FAILED(hr))
 317         {
 318              if (This->pReader)
 319                  IAsyncReader_Release(This->pReader);
 320              This->pReader = NULL;
 321              if (This->prefAlloc)
 322                  IMemAllocator_Release(This->prefAlloc);
 323              This->prefAlloc = NULL;
 324              if (This->pAlloc)
 325                  IMemAllocator_Release(This->pAlloc);
 326              This->pAlloc = NULL;
 327         }
 328     }
 329     else
 330         hr = VFW_E_ALREADY_CONNECTED;
 331     LeaveCriticalSection(This->pin.pCritSec);
 332     return hr;
 333 }
 334
 335 HRESULT WINAPI PullPin_QueryInterface(IPin * iface, REFIID riid, LPVOID * ppv)
 336 {
 337     PullPin *This = impl_PullPin_from_IPin(iface);
 338
 339     TRACE("(%p/%p)->(%s, %p)\n", This, iface, qzdebugstr_guid(riid), ppv);
 340
 341     *ppv = NULL;
 342
 343     if (IsEqualIID(riid, &IID_IUnknown))
 344         *ppv = iface;
 345     else if (IsEqualIID(riid, &IID_IPin))
 346         *ppv = iface;
 347     else if (IsEqualIID(riid, &IID_IMediaSeeking) ||
 348              IsEqualIID(riid, &IID_IQualityControl))
 349     {
 350         return IBaseFilter_QueryInterface(This->pin.pinInfo.pFilter, riid, ppv);
 351     }
 352
 353     if (*ppv)
 354     {
 355         IUnknown_AddRef((IUnknown *)(*ppv));
 356         return S_OK;
 357     }
 358
 359     FIXME("No interface for %s!\n", qzdebugstr_guid(riid));
 360
 361     return E_NOINTERFACE;
 362 }
 363
 364 ULONG WINAPI PullPin_Release(IPin *iface)
 365 {
 366     PullPin *This = impl_PullPin_from_IPin(iface);
 367     ULONG refCount = InterlockedDecrement(&This->pin.refCount);
 368
 369     TRACE("(%p)->() Release from %d\n", This, refCount + 1);
 370
 371     if (!refCount)
 372     {
 373         WaitForSingleObject(This->hEventStateChanged, INFINITE);
 374         assert(!This->hThread);
 375
 376         if(This->prefAlloc)
 377             IMemAllocator_Release(This->prefAlloc);
 378         if(This->pAlloc)
 379             IMemAllocator_Release(This->pAlloc);
 380         if(This->pReader)
 381             IAsyncReader_Release(This->pReader);
 382         CloseHandle(This->thread_sleepy);
 383         CloseHandle(This->hEventStateChanged);
 384         This->thread_lock.DebugInfo->Spare[0] = 0;
 385         DeleteCriticalSection(&This->thread_lock);
 386         CoTaskMemFree(This);
 387         return 0;
 388     }
 389     return refCount;
 390 }
 391
 392 static void PullPin_Flush(PullPin *This)
 393 {
 394     IMediaSample *pSample;
 395     TRACE("Flushing!\n");
 396
 397     if (This->pReader)
 398     {
 399         /* Do not allow state to change while flushing */
 400         EnterCriticalSection(This->pin.pCritSec);
 401
 402         /* Flush outstanding samples */
 403         IAsyncReader_BeginFlush(This->pReader);
 404
 405         for (;;)
 406         {
 407             DWORD_PTR dwUser;
 408
 409             IAsyncReader_WaitForNext(This->pReader, 0, &pSample, &dwUser);
 410
 411             if (!pSample)
 412                 break;
 413
 414             assert(!IMediaSample_GetActualDataLength(pSample));
 415
 416             IMediaSample_Release(pSample);
 417         }
 418
 419         IAsyncReader_EndFlush(This->pReader);
 420
 421         LeaveCriticalSection(This->pin.pCritSec);
 422     }
 423 }
 424
 425 static void PullPin_Thread_Process(PullPin *This)
 426 {
 427     HRESULT hr;
 428     IMediaSample * pSample = NULL;
 429     ALLOCATOR_PROPERTIES allocProps;
 430
 431     hr = IMemAllocator_GetProperties(This->pAlloc, &allocProps);
 432
 433     This->cbAlign = allocProps.cbAlign;
 434
 435     if (This->rtCurrent < This->rtStart)
 436         This->rtCurrent = MEDIATIME_FROM_BYTES(ALIGNDOWN(BYTES_FROM_MEDIATIME(This->rtStart), This->cbAlign));
 437
 438     TRACE("Start\n");
 439
 440     if (This->rtCurrent >= This->rtStop)
 441     {
 442         IPin_EndOfStream(&This->pin.IPin_iface);
 443         return;
 444     }
 445
 446     /* There is no sample in our buffer */
 447     hr = This->fnCustomRequest(This->pUserData);
 448
 449     if (FAILED(hr))
 450         ERR("Request error: %x\n", hr);
 451
 452     EnterCriticalSection(This->pin.pCritSec);
 453     SetEvent(This->hEventStateChanged);
 454     LeaveCriticalSection(This->pin.pCritSec);
 455
 456     if (SUCCEEDED(hr))
 457     do
 458     {
 459         DWORD_PTR dwUser;
 460
 461         TRACE("Process sample\n");
 462
 463         pSample = NULL;
 464         hr = IAsyncReader_WaitForNext(This->pReader, 10000, &pSample, &dwUser);
 465
 466         /* Return an empty sample on error to the implementation in case it does custom parsing, so it knows it's gone */
 467         if (SUCCEEDED(hr))
 468         {
 469             hr = This->fnSampleProc(This->pUserData, pSample, dwUser);
 470         }
 471         else
 472         {
 473             if (hr == VFW_E_TIMEOUT)
 474             {
 475                 if (pSample != NULL)
 476                     WARN("Non-NULL sample returned with VFW_E_TIMEOUT.\n");
 477                 hr = S_OK;
 478             }
 479             /* FIXME: Errors are not well handled yet! */
 480             else
 481                 ERR("Processing error: %x\n", hr);
 482         }
 483
 484         if (pSample)
 485         {
 486             IMediaSample_Release(pSample);
 487             pSample = NULL;
 488         }
 489     } while (This->rtCurrent < This->rtStop && hr == S_OK && !This->stop_playback);
 490
 491     /*
 492      * Sample was rejected, and we are asked to terminate.  When there is more than one buffer
 493      * it is possible for a filter to have several queued samples, making it necessary to
 494      * release all of these pending samples.
 495      */
 496     if (This->stop_playback || FAILED(hr))
 497     {
 498         DWORD_PTR dwUser;
 499
 500         do
 501         {
 502             if (pSample)
 503                 IMediaSample_Release(pSample);
 504             pSample = NULL;
 505             IAsyncReader_WaitForNext(This->pReader, 0, &pSample, &dwUser);
 506         } while(pSample);
 507     }
 508
 509     /* Can't reset state to Sleepy here because that might race, instead PauseProcessing will do that for us
 510      * Flush remaining samples
 511      */
 512     if (This->fnDone)
 513         This->fnDone(This->pUserData);
 514
 515     TRACE("End: %08x, %d\n", hr, This->stop_playback);
 516 }
 517
 518 static void PullPin_Thread_Pause(PullPin *This)
 519 {
 520     PullPin_Flush(This);
 521
 522     EnterCriticalSection(This->pin.pCritSec);
 523     This->state = Req_Sleepy;
 524     SetEvent(This->hEventStateChanged);
 525     LeaveCriticalSection(This->pin.pCritSec);
 526 }
 527
 528 static void  PullPin_Thread_Stop(PullPin *This)
 529 {
 530     TRACE("(%p)->()\n", This);
 531
 532     EnterCriticalSection(This->pin.pCritSec);
 533     {
 534         CloseHandle(This->hThread);
 535         This->hThread = NULL;
 536         SetEvent(This->hEventStateChanged);
 537     }
 538     LeaveCriticalSection(This->pin.pCritSec);
 539
 540     IBaseFilter_Release(This->pin.pinInfo.pFilter);
 541
 542     CoUninitialize();
 543     ExitThread(0);
 544 }
 545
 546 static DWORD WINAPI PullPin_Thread_Main(LPVOID pv)
 547 {
 548     PullPin *This = pv;
 549     CoInitializeEx(NULL, COINIT_MULTITHREADED);
 550
 551     PullPin_Flush(This);
 552
 553     for (;;)
 554     {
 555         WaitForSingleObject(This->thread_sleepy, INFINITE);
 556
 557         TRACE("State: %d\n", This->state);
 558
 559         switch (This->state)
 560         {
 561         case Req_Die: PullPin_Thread_Stop(This); break;
 562         case Req_Run: PullPin_Thread_Process(This); break;
 563         case Req_Pause: PullPin_Thread_Pause(This); break;
 564         case Req_Sleepy: ERR("Should not be signalled with SLEEPY!\n"); break;
 565         default: ERR("Unknown state request: %d\n", This->state); break;
 566         }
 567     }
 568     return 0;
 569 }
 570
 571 static HRESULT PullPin_InitProcessing(PullPin * This)
 572 {
 573     HRESULT hr = S_OK;
 574
 575     TRACE("(%p)->()\n", This);
 576
 577     /* if we are connected */
 578     if (This->pAlloc)
 579     {
 580         DWORD dwThreadId;
 581
 582         WaitForSingleObject(This->hEventStateChanged, INFINITE);
 583         EnterCriticalSection(This->pin.pCritSec);
 584
 585         assert(!This->hThread);
 586         assert(This->state == Req_Die);
 587         assert(This->stop_playback);
 588         assert(WaitForSingleObject(This->thread_sleepy, 0) == WAIT_TIMEOUT);
 589         This->state = Req_Sleepy;
 590
 591         /* AddRef the filter to make sure it and its pins will be around
 592          * as long as the thread */
 593         IBaseFilter_AddRef(This->pin.pinInfo.pFilter);
 594
 595
 596         This->hThread = CreateThread(NULL, 0, PullPin_Thread_Main, This, 0, &dwThreadId);
 597         if (!This->hThread)
 598         {
 599             hr = HRESULT_FROM_WIN32(GetLastError());
 600             IBaseFilter_Release(This->pin.pinInfo.pFilter);
 601         }
 602
 603         if (SUCCEEDED(hr))
 604         {
 605             SetEvent(This->hEventStateChanged);
 606             /* If assert fails, that means a command was not processed before the thread previously terminated */
 607         }
 608         LeaveCriticalSection(This->pin.pCritSec);
 609     }
 610
 611     TRACE(" -- %x\n", hr);
 612
 613     return hr;
 614 }
 615
 616 HRESULT PullPin_StartProcessing(PullPin * This)
 617 {
 618     /* if we are connected */
 619     TRACE("(%p)->()\n", This);
 620     if(This->pAlloc)
 621     {
 622         assert(This->hThread);
 623
 624         PullPin_WaitForStateChange(This, INFINITE);
 625
 626         assert(This->state == Req_Sleepy);
 627
 628         /* Wake up! */
 629         assert(WaitForSingleObject(This->thread_sleepy, 0) == WAIT_TIMEOUT);
 630         This->state = Req_Run;
 631         This->stop_playback = 0;
 632         ResetEvent(This->hEventStateChanged);
 633         SetEvent(This->thread_sleepy);
 634     }
 635
 636     return S_OK;
 637 }
 638
 639 HRESULT PullPin_PauseProcessing(PullPin * This)
 640 {
 641     /* if we are connected */
 642     TRACE("(%p)->()\n", This);
 643     if(This->pAlloc)
 644     {
 645         assert(This->hThread);
 646
 647         PullPin_WaitForStateChange(This, INFINITE);
 648
 649         EnterCriticalSection(This->pin.pCritSec);
 650
 651         assert(!This->stop_playback);
 652         assert(This->state == Req_Run|| This->state == Req_Sleepy);
 653
 654         assert(WaitForSingleObject(This->thread_sleepy, 0) == WAIT_TIMEOUT);
 655
 656         This->state = Req_Pause;
 657         This->stop_playback = 1;
 658         ResetEvent(This->hEventStateChanged);
 659         SetEvent(This->thread_sleepy);
 660
 661         /* Release any outstanding samples */
 662         if (This->pReader)
 663         {
 664             IMediaSample *pSample;
 665             DWORD_PTR dwUser;
 666
 667             do
 668             {
 669                 pSample = NULL;
 670                 IAsyncReader_WaitForNext(This->pReader, 0, &pSample, &dwUser);
 671                 if (pSample)
 672                     IMediaSample_Release(pSample);
 673             } while(pSample);
 674         }
 675
 676         LeaveCriticalSection(This->pin.pCritSec);
 677     }
 678
 679     return S_OK;
 680 }
 681
 682 static HRESULT PullPin_StopProcessing(PullPin * This)
 683 {
 684     TRACE("(%p)->()\n", This);
 685
 686     /* if we are alive */
 687     assert(This->hThread);
 688
 689     PullPin_WaitForStateChange(This, INFINITE);
 690
 691     assert(This->state == Req_Pause || This->state == Req_Sleepy);
 692
 693     This->stop_playback = 1;
 694     This->state = Req_Die;
 695     assert(WaitForSingleObject(This->thread_sleepy, 0) == WAIT_TIMEOUT);
 696     ResetEvent(This->hEventStateChanged);
 697     SetEvent(This->thread_sleepy);
 698     return S_OK;
 699 }
 700
 701 HRESULT PullPin_WaitForStateChange(PullPin * This, DWORD dwMilliseconds)
 702 {
 703     if (WaitForSingleObject(This->hEventStateChanged, dwMilliseconds) == WAIT_TIMEOUT)
 704         return S_FALSE;
 705     return S_OK;
 706 }
 707
 708 HRESULT WINAPI PullPin_QueryAccept(IPin * iface, const AM_MEDIA_TYPE * pmt)
 709 {
 710     PullPin *This = impl_PullPin_from_IPin(iface);
 711
 712     TRACE("(%p/%p)->(%p)\n", This, iface, pmt);
 713
 714     return (This->fnQueryAccept(This->pUserData, pmt) == S_OK ? S_OK : S_FALSE);
 715 }
 716
 717 HRESULT WINAPI PullPin_EndOfStream(IPin * iface)
 718 {
 719     PullPin *This = impl_PullPin_from_IPin(iface);
 720     HRESULT hr = S_FALSE;
 721
 722     TRACE("(%p)->()\n", iface);
 723
 724     EnterCriticalSection(This->pin.pCritSec);
 725     hr = SendFurther( iface, deliver_endofstream, NULL, NULL );
 726     SetEvent(This->hEventStateChanged);
 727     LeaveCriticalSection(This->pin.pCritSec);
 728
 729     return hr;
 730 }
 731
 732 HRESULT WINAPI PullPin_BeginFlush(IPin * iface)
 733 {
 734     PullPin *This = impl_PullPin_from_IPin(iface);
 735     TRACE("(%p)->()\n", This);
 736
 737     EnterCriticalSection(This->pin.pCritSec);
 738     {
 739         SendFurther( iface, deliver_beginflush, NULL, NULL );
 740     }
 741     LeaveCriticalSection(This->pin.pCritSec);
 742
 743     EnterCriticalSection(&This->thread_lock);
 744     {
 745         if (This->pReader)
 746             IAsyncReader_BeginFlush(This->pReader);
 747         PullPin_WaitForStateChange(This, INFINITE);
 748
 749         if (This->hThread && This->state == Req_Run)
 750         {
 751             PullPin_PauseProcessing(This);
 752             PullPin_WaitForStateChange(This, INFINITE);
 753         }
 754     }
 755     LeaveCriticalSection(&This->thread_lock);
 756
 757     EnterCriticalSection(This->pin.pCritSec);
 758     {
 759         This->fnCleanProc(This->pUserData);
 760     }
 761     LeaveCriticalSection(This->pin.pCritSec);
 762
 763     return S_OK;
 764 }
 765
 766 HRESULT WINAPI PullPin_EndFlush(IPin * iface)
 767 {
 768     PullPin *This = impl_PullPin_from_IPin(iface);
 769
 770     TRACE("(%p)->()\n", iface);
 771
 772     /* Send further first: Else a race condition might terminate processing early */
 773     EnterCriticalSection(This->pin.pCritSec);
 774     SendFurther( iface, deliver_endflush, NULL, NULL );
 775     LeaveCriticalSection(This->pin.pCritSec);
 776
 777     EnterCriticalSection(&This->thread_lock);
 778     {
 779         FILTER_STATE state;
 780
 781         if (This->pReader)
 782             IAsyncReader_EndFlush(This->pReader);
 783
 784         IBaseFilter_GetState(This->pin.pinInfo.pFilter, INFINITE, &state);
 785
 786         if (state != State_Stopped)
 787             PullPin_StartProcessing(This);
 788
 789         PullPin_WaitForStateChange(This, INFINITE);
 790     }
 791     LeaveCriticalSection(&This->thread_lock);
 792
 793     return S_OK;
 794 }
 795
 796 HRESULT WINAPI PullPin_Disconnect(IPin *iface)
 797 {
 798     HRESULT hr;
 799     PullPin *This = impl_PullPin_from_IPin(iface);
 800
 801     TRACE("()\n");
 802
 803     EnterCriticalSection(This->pin.pCritSec);
 804     {
 805         if (FAILED(hr = IMemAllocator_Decommit(This->pAlloc)))
 806             ERR("Allocator decommit failed with error %x. Possible memory leak\n", hr);
 807
 808         if (This->pin.pConnectedTo)
 809         {
 810             IPin_Release(This->pin.pConnectedTo);
 811             This->pin.pConnectedTo = NULL;
 812             PullPin_StopProcessing(This);
 813
 814             FreeMediaType(&This->pin.mtCurrent);
 815             ZeroMemory(&This->pin.mtCurrent, sizeof(This->pin.mtCurrent));
 816             hr = S_OK;
 817         }
 818         else
 819             hr = S_FALSE;
 820     }
 821     LeaveCriticalSection(This->pin.pCritSec);
 822
 823     return hr;
 824 }
 825
 826 HRESULT WINAPI PullPin_NewSegment(IPin * iface, REFERENCE_TIME tStart, REFERENCE_TIME tStop, double dRate)
 827 {
 828     newsegmentargs args;
 829     FIXME("(%p)->(%s, %s, %g) stub\n", iface, wine_dbgstr_longlong(tStart), wine_dbgstr_longlong(tStop), dRate);
 830
 831     args.tStart = tStart;
 832     args.tStop = tStop;
 833     args.rate = dRate;
 834
 835     return SendFurther( iface, deliver_newsegment, &args, NULL );
 836 }
 837
 838 static const IPinVtbl PullPin_Vtbl =
 839 {
 840     PullPin_QueryInterface,
 841     BasePinImpl_AddRef,
 842     PullPin_Release,
 843     BaseInputPinImpl_Connect,
 844     PullPin_ReceiveConnection,
 845     PullPin_Disconnect,
 846     BasePinImpl_ConnectedTo,
 847     BasePinImpl_ConnectionMediaType,
 848     BasePinImpl_QueryPinInfo,
 849     BasePinImpl_QueryDirection,
 850     BasePinImpl_QueryId,
 851     PullPin_QueryAccept,
 852     BasePinImpl_EnumMediaTypes,
 853     BasePinImpl_QueryInternalConnections,
 854     PullPin_EndOfStream,
 855     PullPin_BeginFlush,
 856     PullPin_EndFlush,
 857     PullPin_NewSegment
 858 };