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