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