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 #define ALIGNDOWN(value,boundary) ((value)/(boundary)*(boundary))
  33 #define ALIGNUP(value,boundary) (ALIGNDOWN((value)+(boundary)-1, (boundary)))
  34
  35 typedef HRESULT (*SendPinFunc)( IPin *to, LPVOID arg );
  36
  37 /** Helper function, there are a lot of places where the error code is inherited
  38  * The following rules apply:
  39  *
  40  * Return the first received error code (E_NOTIMPL is ignored)
  41  * If no errors occur: return the first received non-error-code that isn't S_OK
  42  */
  43 static HRESULT updatehres( HRESULT original, HRESULT new )
  44 {
  45     if (FAILED( original ) || new == E_NOTIMPL)
  46         return original;
  47
  48     if (FAILED( new ) || original == S_OK)
  49         return new;
  50
  51     return original;
  52 }
  53
  54 /** Sends a message from a pin further to other, similar pins
  55  * fnMiddle is called on each pin found further on the stream.
  56  * fnEnd (can be NULL) is called when the message can't be sent any further (this is a renderer or source)
  57  *
  58  * If the pin given is an input pin, the message will be sent downstream to other input pins
  59  * If the pin given is an output pin, the message will be sent upstream to other output pins
  60  */
  61 static HRESULT SendFurther( IPin *from, SendPinFunc fnMiddle, LPVOID arg, SendPinFunc fnEnd )
  62 {
  63     PIN_INFO pin_info;
  64     ULONG amount = 0;
  65     HRESULT hr = S_OK;
  66     HRESULT hr_return = S_OK;
  67     IEnumPins *enumpins = NULL;
  68     BOOL foundend = TRUE;
  69     PIN_DIRECTION from_dir;
  70
  71     IPin_QueryDirection( from, &from_dir );
  72
  73     hr = IPin_QueryInternalConnections( from, NULL, &amount );
  74     if (hr != E_NOTIMPL && amount)
  75         FIXME("Use QueryInternalConnections!\n");
  76
  77     pin_info.pFilter = NULL;
  78     hr = IPin_QueryPinInfo( from, &pin_info );
  79     if (FAILED(hr))
  80         goto out;
  81
  82     hr = IBaseFilter_EnumPins( pin_info.pFilter, &enumpins );
  83     if (FAILED(hr))
  84         goto out;
  85
  86     hr = IEnumPins_Reset( enumpins );
  87     while (hr == S_OK) {
  88         IPin *pin = NULL;
  89         hr = IEnumPins_Next( enumpins, 1, &pin, NULL );
  90         if (hr == VFW_E_ENUM_OUT_OF_SYNC)
  91         {
  92             hr = IEnumPins_Reset( enumpins );
  93             continue;
  94         }
  95         if (pin)
  96         {
  97             PIN_DIRECTION dir;
  98
  99             IPin_QueryDirection( pin, &dir );
 100             if (dir != from_dir)
 101             {
 102                 IPin *connected = NULL;
 103
 104                 foundend = FALSE;
 105                 IPin_ConnectedTo( pin, &connected );
 106                 if (connected)
 107                 {
 108                     HRESULT hr_local;
 109
 110                     hr_local = fnMiddle( connected, arg );
 111                     hr_return = updatehres( hr_return, hr_local );
 112                     IPin_Release(connected);
 113                 }
 114             }
 115             IPin_Release( pin );
 116         }
 117         else
 118         {
 119             hr = S_OK;
 120             break;
 121         }
 122     }
 123
 124     if (!foundend)
 125         hr = hr_return;
 126     else if (fnEnd) {
 127         HRESULT hr_local;
 128
 129         hr_local = fnEnd( from, arg );
 130         hr_return = updatehres( hr_return, hr_local );
 131     }
 132
 133 out:
 134     if (enumpins)
 135         IEnumPins_Release( enumpins );
 136     if (pin_info.pFilter)
 137         IBaseFilter_Release( pin_info.pFilter );
 138     return hr;
 139 }
 140
 141
 142 static void Copy_PinInfo(PIN_INFO * pDest, const PIN_INFO * pSrc)
 143 {
 144     /* Tempting to just do a memcpy, but the name field is
 145        128 characters long! We will probably never exceed 10
 146        most of the time, so we are better off copying
 147        each field manually */
 148     strcpyW(pDest->achName, pSrc->achName);
 149     pDest->dir = pSrc->dir;
 150     pDest->pFilter = pSrc->pFilter;
 151 }
 152
 153 static HRESULT deliver_endofstream(IPin* pin, LPVOID unused)
 154 {
 155     return IPin_EndOfStream( pin );
 156 }
 157
 158 static HRESULT deliver_beginflush(IPin* pin, LPVOID unused)
 159 {
 160     return IPin_BeginFlush( pin );
 161 }
 162
 163 static HRESULT deliver_endflush(IPin* pin, LPVOID unused)
 164 {
 165     return IPin_EndFlush( pin );
 166 }
 167
 168 typedef struct newsegmentargs
 169 {
 170     REFERENCE_TIME tStart, tStop;
 171     double rate;
 172 } newsegmentargs;
 173
 174 static HRESULT deliver_newsegment(IPin *pin, LPVOID data)
 175 {
 176     newsegmentargs *args = data;
 177     return IPin_NewSegment(pin, args->tStart, args->tStop, args->rate);
 178 }
 179
 180 /*** PullPin implementation ***/
 181
 182 static HRESULT PullPin_Init(const IPinVtbl *PullPin_Vtbl, const PIN_INFO * pPinInfo, SAMPLEPROC_PULL pSampleProc, LPVOID pUserData,
 183                             QUERYACCEPTPROC pQueryAccept, CLEANUPPROC pCleanUp, REQUESTPROC pCustomRequest, STOPPROCESSPROC pDone, LPCRITICAL_SECTION pCritSec, PullPin * pPinImpl)
 184 {
 185     /* Common attributes */
 186     pPinImpl->pin.IPin_iface.lpVtbl = PullPin_Vtbl;
 187     pPinImpl->pin.refCount = 1;
 188     pPinImpl->pin.pConnectedTo = NULL;
 189     pPinImpl->pin.pCritSec = pCritSec;
 190     Copy_PinInfo(&pPinImpl->pin.pinInfo, pPinInfo);
 191     ZeroMemory(&pPinImpl->pin.mtCurrent, sizeof(AM_MEDIA_TYPE));
 192
 193     /* Input pin attributes */
 194     pPinImpl->pUserData = pUserData;
 195     pPinImpl->fnQueryAccept = pQueryAccept;
 196     pPinImpl->fnSampleProc = pSampleProc;
 197     pPinImpl->fnCleanProc = pCleanUp;
 198     pPinImpl->fnDone = pDone;
 199     pPinImpl->fnPreConnect = NULL;
 200     pPinImpl->pAlloc = NULL;
 201     pPinImpl->prefAlloc = 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 = TRUE;
 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 = &pPinImpl->pin.IPin_iface;
 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 = impl_PullPin_from_IPin(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; /* 64 KB */
 266         props.cbAlign = 1;
 267         props.cbPrefix = 0;
 268
 269         if (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         This->prefAlloc = NULL;
 287         if (SUCCEEDED(hr))
 288         {
 289             hr = IPin_QueryInterface(pReceivePin, &IID_IAsyncReader, (LPVOID *)&This->pReader);
 290         }
 291
 292         if (SUCCEEDED(hr) && This->fnPreConnect)
 293         {
 294             hr = This->fnPreConnect(iface, pReceivePin, &props);
 295         }
 296
 297         /*
 298          * Some custom filters (such as the one used by Fallout 3
 299          * and Fallout: New Vegas) expect to be passed a non-NULL
 300          * preferred allocator.
 301          */
 302         if (SUCCEEDED(hr))
 303         {
 304             hr = StdMemAllocator_create(NULL, (LPVOID *) &This->prefAlloc);
 305         }
 306
 307         if (SUCCEEDED(hr))
 308         {
 309             hr = IAsyncReader_RequestAllocator(This->pReader, This->prefAlloc, &props, &This->pAlloc);
 310         }
 311
 312         if (SUCCEEDED(hr))
 313         {
 314             CopyMediaType(&This->pin.mtCurrent, pmt);
 315             This->pin.pConnectedTo = pReceivePin;
 316             IPin_AddRef(pReceivePin);
 317             hr = IMemAllocator_Commit(This->pAlloc);
 318         }
 319
 320         if (SUCCEEDED(hr))
 321             hr = PullPin_InitProcessing(This);
 322
 323         if (FAILED(hr))
 324         {
 325              if (This->pReader)
 326                  IAsyncReader_Release(This->pReader);
 327              This->pReader = NULL;
 328              if (This->prefAlloc)
 329                  IMemAllocator_Release(This->prefAlloc);
 330              This->prefAlloc = NULL;
 331              if (This->pAlloc)
 332                  IMemAllocator_Release(This->pAlloc);
 333              This->pAlloc = NULL;
 334         }
 335     }
 336     else
 337         hr = VFW_E_ALREADY_CONNECTED;
 338     LeaveCriticalSection(This->pin.pCritSec);
 339     return hr;
 340 }
 341
 342 HRESULT WINAPI PullPin_QueryInterface(IPin * iface, REFIID riid, LPVOID * ppv)
 343 {
 344     PullPin *This = impl_PullPin_from_IPin(iface);
 345
 346     TRACE("(%p/%p)->(%s, %p)\n", This, iface, qzdebugstr_guid(riid), ppv);
 347
 348     *ppv = NULL;
 349
 350     if (IsEqualIID(riid, &IID_IUnknown))
 351         *ppv = iface;
 352     else if (IsEqualIID(riid, &IID_IPin))
 353         *ppv = iface;
 354     else if (IsEqualIID(riid, &IID_IMediaSeeking) ||
 355              IsEqualIID(riid, &IID_IQualityControl))
 356     {
 357         return IBaseFilter_QueryInterface(This->pin.pinInfo.pFilter, riid, ppv);
 358     }
 359
 360     if (*ppv)
 361     {
 362         IUnknown_AddRef((IUnknown *)(*ppv));
 363         return S_OK;
 364     }
 365
 366     FIXME("No interface for %s!\n", qzdebugstr_guid(riid));
 367
 368     return E_NOINTERFACE;
 369 }
 370
 371 ULONG WINAPI PullPin_Release(IPin *iface)
 372 {
 373     PullPin *This = impl_PullPin_from_IPin(iface);
 374     ULONG refCount = InterlockedDecrement(&This->pin.refCount);
 375
 376     TRACE("(%p)->() Release from %d\n", This, refCount + 1);
 377
 378     if (!refCount)
 379     {
 380         WaitForSingleObject(This->hEventStateChanged, INFINITE);
 381         assert(!This->hThread);
 382
 383         if(This->prefAlloc)
 384             IMemAllocator_Release(This->prefAlloc);
 385         if(This->pAlloc)
 386             IMemAllocator_Release(This->pAlloc);
 387         if(This->pReader)
 388             IAsyncReader_Release(This->pReader);
 389         CloseHandle(This->thread_sleepy);
 390         CloseHandle(This->hEventStateChanged);
 391         This->thread_lock.DebugInfo->Spare[0] = 0;
 392         DeleteCriticalSection(&This->thread_lock);
 393         CoTaskMemFree(This);
 394         return 0;
 395     }
 396     return refCount;
 397 }
 398
 399 static void PullPin_Flush(PullPin *This)
 400 {
 401     IMediaSample *pSample;
 402     TRACE("Flushing!\n");
 403
 404     if (This->pReader)
 405     {
 406         /* Do not allow state to change while flushing */
 407         EnterCriticalSection(This->pin.pCritSec);
 408
 409         /* Flush outstanding samples */
 410         IAsyncReader_BeginFlush(This->pReader);
 411
 412         for (;;)
 413         {
 414             DWORD_PTR dwUser;
 415
 416             pSample = NULL;
 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(&This->pin.IPin_iface);
 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             if (hr == VFW_E_TIMEOUT)
 482             {
 483                 if (pSample != NULL)
 484                     WARN("Non-NULL sample returned with VFW_E_TIMEOUT.\n");
 485                 hr = S_OK;
 486             }
 487             /* FIXME: Errors are not well handled yet! */
 488             else
 489                 ERR("Processing error: %x\n", hr);
 490         }
 491
 492         if (pSample)
 493         {
 494             IMediaSample_Release(pSample);
 495             pSample = NULL;
 496         }
 497     } while (This->rtCurrent < This->rtStop && hr == S_OK && !This->stop_playback);
 498
 499     /*
 500      * Sample was rejected, and we are asked to terminate.  When there is more than one buffer
 501      * it is possible for a filter to have several queued samples, making it necessary to
 502      * release all of these pending samples.
 503      */
 504     if (This->stop_playback || FAILED(hr))
 505     {
 506         DWORD_PTR dwUser;
 507
 508         do
 509         {
 510             if (pSample)
 511                 IMediaSample_Release(pSample);
 512             pSample = NULL;
 513             IAsyncReader_WaitForNext(This->pReader, 0, &pSample, &dwUser);
 514         } while(pSample);
 515     }
 516
 517     /* Can't reset state to Sleepy here because that might race, instead PauseProcessing will do that for us
 518      * Flush remaining samples
 519      */
 520     if (This->fnDone)
 521         This->fnDone(This->pUserData);
 522
 523     TRACE("End: %08x, %d\n", hr, This->stop_playback);
 524 }
 525
 526 static void PullPin_Thread_Pause(PullPin *This)
 527 {
 528     PullPin_Flush(This);
 529
 530     EnterCriticalSection(This->pin.pCritSec);
 531     This->state = Req_Sleepy;
 532     SetEvent(This->hEventStateChanged);
 533     LeaveCriticalSection(This->pin.pCritSec);
 534 }
 535
 536 static void  PullPin_Thread_Stop(PullPin *This)
 537 {
 538     TRACE("(%p)->()\n", This);
 539
 540     EnterCriticalSection(This->pin.pCritSec);
 541     {
 542         CloseHandle(This->hThread);
 543         This->hThread = NULL;
 544         SetEvent(This->hEventStateChanged);
 545     }
 546     LeaveCriticalSection(This->pin.pCritSec);
 547
 548     IBaseFilter_Release(This->pin.pinInfo.pFilter);
 549
 550     CoUninitialize();
 551     ExitThread(0);
 552 }
 553
 554 static DWORD WINAPI PullPin_Thread_Main(LPVOID pv)
 555 {
 556     PullPin *This = pv;
 557     CoInitializeEx(NULL, COINIT_MULTITHREADED);
 558
 559     PullPin_Flush(This);
 560
 561     for (;;)
 562     {
 563         WaitForSingleObject(This->thread_sleepy, INFINITE);
 564
 565         TRACE("State: %d\n", This->state);
 566
 567         switch (This->state)
 568         {
 569         case Req_Die: PullPin_Thread_Stop(This); break;
 570         case Req_Run: PullPin_Thread_Process(This); break;
 571         case Req_Pause: PullPin_Thread_Pause(This); break;
 572         case Req_Sleepy: ERR("Should not be signalled with SLEEPY!\n"); break;
 573         default: ERR("Unknown state request: %d\n", This->state); break;
 574         }
 575     }
 576     return 0;
 577 }
 578
 579 static HRESULT PullPin_InitProcessing(PullPin * This)
 580 {
 581     HRESULT hr = S_OK;
 582
 583     TRACE("(%p)->()\n", This);
 584
 585     /* if we are connected */
 586     if (This->pAlloc)
 587     {
 588         DWORD dwThreadId;
 589
 590         WaitForSingleObject(This->hEventStateChanged, INFINITE);
 591         EnterCriticalSection(This->pin.pCritSec);
 592
 593         assert(!This->hThread);
 594         assert(This->state == Req_Die);
 595         assert(This->stop_playback);
 596         assert(WaitForSingleObject(This->thread_sleepy, 0) == WAIT_TIMEOUT);
 597         This->state = Req_Sleepy;
 598
 599         /* AddRef the filter to make sure it and its pins will be around
 600          * as long as the thread */
 601         IBaseFilter_AddRef(This->pin.pinInfo.pFilter);
 602
 603
 604         This->hThread = CreateThread(NULL, 0, PullPin_Thread_Main, This, 0, &dwThreadId);
 605         if (!This->hThread)
 606         {
 607             hr = HRESULT_FROM_WIN32(GetLastError());
 608             IBaseFilter_Release(This->pin.pinInfo.pFilter);
 609         }
 610
 611         if (SUCCEEDED(hr))
 612         {
 613             SetEvent(This->hEventStateChanged);
 614             /* If assert fails, that means a command was not processed before the thread previously terminated */
 615         }
 616         LeaveCriticalSection(This->pin.pCritSec);
 617     }
 618
 619     TRACE(" -- %x\n", hr);
 620
 621     return hr;
 622 }
 623
 624 HRESULT PullPin_StartProcessing(PullPin * This)
 625 {
 626     /* if we are connected */
 627     TRACE("(%p)->()\n", This);
 628     if(This->pAlloc)
 629     {
 630         assert(This->hThread);
 631
 632         PullPin_WaitForStateChange(This, INFINITE);
 633
 634         assert(This->state == Req_Sleepy);
 635
 636         /* Wake up! */
 637         assert(WaitForSingleObject(This->thread_sleepy, 0) == WAIT_TIMEOUT);
 638         This->state = Req_Run;
 639         This->stop_playback = FALSE;
 640         ResetEvent(This->hEventStateChanged);
 641         SetEvent(This->thread_sleepy);
 642     }
 643
 644     return S_OK;
 645 }
 646
 647 HRESULT PullPin_PauseProcessing(PullPin * This)
 648 {
 649     /* if we are connected */
 650     TRACE("(%p)->()\n", This);
 651     if(This->pAlloc)
 652     {
 653         assert(This->hThread);
 654
 655         PullPin_WaitForStateChange(This, INFINITE);
 656
 657         EnterCriticalSection(This->pin.pCritSec);
 658
 659         assert(!This->stop_playback);
 660         assert(This->state == Req_Run|| This->state == Req_Sleepy);
 661
 662         assert(WaitForSingleObject(This->thread_sleepy, 0) == WAIT_TIMEOUT);
 663
 664         This->state = Req_Pause;
 665         This->stop_playback = TRUE;
 666         ResetEvent(This->hEventStateChanged);
 667         SetEvent(This->thread_sleepy);
 668
 669         /* Release any outstanding samples */
 670         if (This->pReader)
 671         {
 672             IMediaSample *pSample;
 673             DWORD_PTR dwUser;
 674
 675             do
 676             {
 677                 pSample = NULL;
 678                 IAsyncReader_WaitForNext(This->pReader, 0, &pSample, &dwUser);
 679                 if (pSample)
 680                     IMediaSample_Release(pSample);
 681             } while(pSample);
 682         }
 683
 684         LeaveCriticalSection(This->pin.pCritSec);
 685     }
 686
 687     return S_OK;
 688 }
 689
 690 static HRESULT PullPin_StopProcessing(PullPin * This)
 691 {
 692     TRACE("(%p)->()\n", This);
 693
 694     /* if we are alive */
 695     assert(This->hThread);
 696
 697     PullPin_WaitForStateChange(This, INFINITE);
 698
 699     assert(This->state == Req_Pause || This->state == Req_Sleepy);
 700
 701     This->stop_playback = TRUE;
 702     This->state = Req_Die;
 703     assert(WaitForSingleObject(This->thread_sleepy, 0) == WAIT_TIMEOUT);
 704     ResetEvent(This->hEventStateChanged);
 705     SetEvent(This->thread_sleepy);
 706     return S_OK;
 707 }
 708
 709 HRESULT PullPin_WaitForStateChange(PullPin * This, DWORD dwMilliseconds)
 710 {
 711     if (WaitForSingleObject(This->hEventStateChanged, dwMilliseconds) == WAIT_TIMEOUT)
 712         return S_FALSE;
 713     return S_OK;
 714 }
 715
 716 HRESULT WINAPI PullPin_QueryAccept(IPin * iface, const AM_MEDIA_TYPE * pmt)
 717 {
 718     PullPin *This = impl_PullPin_from_IPin(iface);
 719
 720     TRACE("(%p/%p)->(%p)\n", This, iface, pmt);
 721
 722     return (This->fnQueryAccept(This->pUserData, pmt) == S_OK ? S_OK : S_FALSE);
 723 }
 724
 725 HRESULT WINAPI PullPin_EndOfStream(IPin * iface)
 726 {
 727     PullPin *This = impl_PullPin_from_IPin(iface);
 728     HRESULT hr = S_FALSE;
 729
 730     TRACE("(%p)->()\n", iface);
 731
 732     EnterCriticalSection(This->pin.pCritSec);
 733     hr = SendFurther( iface, deliver_endofstream, NULL, NULL );
 734     SetEvent(This->hEventStateChanged);
 735     LeaveCriticalSection(This->pin.pCritSec);
 736
 737     return hr;
 738 }
 739
 740 HRESULT WINAPI PullPin_BeginFlush(IPin * iface)
 741 {
 742     PullPin *This = impl_PullPin_from_IPin(iface);
 743     TRACE("(%p)->()\n", This);
 744
 745     EnterCriticalSection(This->pin.pCritSec);
 746     {
 747         SendFurther( iface, deliver_beginflush, NULL, NULL );
 748     }
 749     LeaveCriticalSection(This->pin.pCritSec);
 750
 751     EnterCriticalSection(&This->thread_lock);
 752     {
 753         if (This->pReader)
 754             IAsyncReader_BeginFlush(This->pReader);
 755         PullPin_WaitForStateChange(This, INFINITE);
 756
 757         if (This->hThread && This->state == Req_Run)
 758         {
 759             PullPin_PauseProcessing(This);
 760             PullPin_WaitForStateChange(This, INFINITE);
 761         }
 762     }
 763     LeaveCriticalSection(&This->thread_lock);
 764
 765     EnterCriticalSection(This->pin.pCritSec);
 766     {
 767         This->fnCleanProc(This->pUserData);
 768     }
 769     LeaveCriticalSection(This->pin.pCritSec);
 770
 771     return S_OK;
 772 }
 773
 774 HRESULT WINAPI PullPin_EndFlush(IPin * iface)
 775 {
 776     PullPin *This = impl_PullPin_from_IPin(iface);
 777
 778     TRACE("(%p)->()\n", iface);
 779
 780     /* Send further first: Else a race condition might terminate processing early */
 781     EnterCriticalSection(This->pin.pCritSec);
 782     SendFurther( iface, deliver_endflush, NULL, NULL );
 783     LeaveCriticalSection(This->pin.pCritSec);
 784
 785     EnterCriticalSection(&This->thread_lock);
 786     {
 787         FILTER_STATE state;
 788
 789         if (This->pReader)
 790             IAsyncReader_EndFlush(This->pReader);
 791
 792         IBaseFilter_GetState(This->pin.pinInfo.pFilter, INFINITE, &state);
 793
 794         if (state != State_Stopped)
 795             PullPin_StartProcessing(This);
 796
 797         PullPin_WaitForStateChange(This, INFINITE);
 798     }
 799     LeaveCriticalSection(&This->thread_lock);
 800
 801     return S_OK;
 802 }
 803
 804 HRESULT WINAPI PullPin_Disconnect(IPin *iface)
 805 {
 806     HRESULT hr;
 807     PullPin *This = impl_PullPin_from_IPin(iface);
 808
 809     TRACE("()\n");
 810
 811     EnterCriticalSection(This->pin.pCritSec);
 812     {
 813         if (FAILED(hr = IMemAllocator_Decommit(This->pAlloc)))
 814             ERR("Allocator decommit failed with error %x. Possible memory leak\n", hr);
 815
 816         if (This->pin.pConnectedTo)
 817         {
 818             IPin_Release(This->pin.pConnectedTo);
 819             This->pin.pConnectedTo = NULL;
 820             PullPin_StopProcessing(This);
 821
 822             FreeMediaType(&This->pin.mtCurrent);
 823             ZeroMemory(&This->pin.mtCurrent, sizeof(This->pin.mtCurrent));
 824             hr = S_OK;
 825         }
 826         else
 827             hr = S_FALSE;
 828     }
 829     LeaveCriticalSection(This->pin.pCritSec);
 830
 831     return hr;
 832 }
 833
 834 HRESULT WINAPI PullPin_NewSegment(IPin * iface, REFERENCE_TIME tStart, REFERENCE_TIME tStop, double dRate)
 835 {
 836     newsegmentargs args;
 837     FIXME("(%p)->(%s, %s, %g) stub\n", iface, wine_dbgstr_longlong(tStart), wine_dbgstr_longlong(tStop), dRate);
 838
 839     args.tStart = tStart;
 840     args.tStop = tStop;
 841     args.rate = dRate;
 842
 843     return SendFurther( iface, deliver_newsegment, &args, NULL );
 844 }