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