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