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