2 * Generic Implementation of IPin Interface
4 * Copyright 2003 Robert Shearman
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.
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.
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
21 #include "quartz_private.h"
24 #include "wine/debug.h"
25 #include "wine/unicode.h"
30 WINE_DEFAULT_DEBUG_CHANNEL(quartz);
32 static const IPinVtbl PullPin_Vtbl;
34 #define ALIGNDOWN(value,boundary) ((value)/(boundary)*(boundary))
35 #define ALIGNUP(value,boundary) (ALIGNDOWN((value)+(boundary)-1, (boundary)))
37 typedef HRESULT (*SendPinFunc)( IPin *to, LPVOID arg );
39 /** Helper function, there are a lot of places where the error code is inherited
40 * The following rules apply:
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
45 HRESULT updatehres( HRESULT original, HRESULT new )
47 if (FAILED( original ) || new == E_NOTIMPL)
50 if (FAILED( new ) || original == S_OK)
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)
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
63 static HRESULT SendFurther( IPin *from, SendPinFunc fnMiddle, LPVOID arg, SendPinFunc fnEnd )
68 HRESULT hr_return = S_OK;
69 IEnumPins *enumpins = NULL;
71 PIN_DIRECTION from_dir;
73 IPin_QueryDirection( from, &from_dir );
75 hr = IPin_QueryInternalConnections( from, NULL, &amount );
76 if (hr != E_NOTIMPL && amount)
77 FIXME("Use QueryInternalConnections!\n");
80 pin_info.pFilter = NULL;
81 hr = IPin_QueryPinInfo( from, &pin_info );
85 hr = IBaseFilter_EnumPins( pin_info.pFilter, &enumpins );
89 hr = IEnumPins_Reset( enumpins );
92 hr = IEnumPins_Next( enumpins, 1, &pin, NULL );
93 if (hr == VFW_E_ENUM_OUT_OF_SYNC)
95 hr = IEnumPins_Reset( enumpins );
102 IPin_QueryDirection( pin, &dir );
105 IPin *connected = NULL;
108 IPin_ConnectedTo( pin, &connected );
113 hr_local = fnMiddle( connected, arg );
114 hr_return = updatehres( hr_return, hr_local );
115 IPin_Release(connected);
132 hr_local = fnEnd( from, arg );
133 hr_return = updatehres( hr_return, hr_local );
137 if (pin_info.pFilter)
138 IBaseFilter_Release( pin_info.pFilter );
143 static void Copy_PinInfo(PIN_INFO * pDest, const PIN_INFO * pSrc)
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;
154 static HRESULT deliver_endofstream(IPin* pin, LPVOID unused)
156 return IPin_EndOfStream( pin );
159 static HRESULT deliver_beginflush(IPin* pin, LPVOID unused)
161 return IPin_BeginFlush( pin );
164 static HRESULT deliver_endflush(IPin* pin, LPVOID unused)
166 return IPin_EndFlush( pin );
169 typedef struct newsegmentargs
171 REFERENCE_TIME tStart, tStop;
175 static HRESULT deliver_newsegment(IPin *pin, LPVOID data)
177 newsegmentargs *args = data;
178 return IPin_NewSegment(pin, args->tStart, args->tStop, args->rate);
181 /*** PullPin implementation ***/
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)
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));
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);
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;
215 InitializeCriticalSection(&pPinImpl->thread_lock);
216 pPinImpl->thread_lock.DebugInfo->Spare[0] = (DWORD_PTR)( __FILE__ ": PullPin.thread_lock");
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)
227 if (pPinInfo->dir != PINDIR_INPUT)
229 ERR("Pin direction(%x) != PINDIR_INPUT\n", pPinInfo->dir);
233 pPinImpl = CoTaskMemAlloc(sizeof(*pPinImpl));
236 return E_OUTOFMEMORY;
238 if (SUCCEEDED(PullPin_Init(PullPin_Vtbl, pPinInfo, pSampleProc, pUserData, pQueryAccept, pCleanUp, pCustomRequest, pDone, pCritSec, pPinImpl)))
240 *ppPin = (IPin *)(&pPinImpl->pin.lpVtbl);
244 CoTaskMemFree(pPinImpl);
248 static HRESULT PullPin_InitProcessing(PullPin * This);
250 HRESULT WINAPI PullPin_ReceiveConnection(IPin * iface, IPin * pReceivePin, const AM_MEDIA_TYPE * pmt)
252 PIN_DIRECTION pindirReceive;
254 PullPin *This = (PullPin *)iface;
256 TRACE("(%p/%p)->(%p, %p)\n", This, iface, pReceivePin, pmt);
257 dump_AM_MEDIA_TYPE(pmt);
259 EnterCriticalSection(This->pin.pCritSec);
260 if (!This->pin.pConnectedTo)
262 ALLOCATOR_PROPERTIES props;
265 props.cbBuffer = 64 * 1024; /* 64k bytes */
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? */
275 IPin_QueryDirection(pReceivePin, &pindirReceive);
277 if (pindirReceive != PINDIR_OUTPUT)
279 ERR("Can't connect from non-output pin\n");
280 hr = VFW_E_INVALID_DIRECTION;
284 This->pReader = NULL;
288 hr = IPin_QueryInterface(pReceivePin, &IID_IAsyncReader, (LPVOID *)&This->pReader);
291 if (SUCCEEDED(hr) && This->fnPreConnect)
293 hr = This->fnPreConnect(iface, pReceivePin, &props);
298 hr = IAsyncReader_RequestAllocator(This->pReader, NULL, &props, &This->pAlloc);
303 CopyMediaType(&This->pin.mtCurrent, pmt);
304 This->pin.pConnectedTo = pReceivePin;
305 IPin_AddRef(pReceivePin);
306 hr = IMemAllocator_Commit(This->pAlloc);
310 hr = PullPin_InitProcessing(This);
315 IAsyncReader_Release(This->pReader);
316 This->pReader = NULL;
318 IMemAllocator_Release(This->pAlloc);
323 hr = VFW_E_ALREADY_CONNECTED;
324 LeaveCriticalSection(This->pin.pCritSec);
328 HRESULT WINAPI PullPin_QueryInterface(IPin * iface, REFIID riid, LPVOID * ppv)
330 PullPin *This = (PullPin *)iface;
332 TRACE("(%p/%p)->(%s, %p)\n", This, iface, qzdebugstr_guid(riid), ppv);
336 if (IsEqualIID(riid, &IID_IUnknown))
338 else if (IsEqualIID(riid, &IID_IPin))
340 else if (IsEqualIID(riid, &IID_IMediaSeeking))
342 return IBaseFilter_QueryInterface(This->pin.pinInfo.pFilter, &IID_IMediaSeeking, ppv);
347 IUnknown_AddRef((IUnknown *)(*ppv));
351 FIXME("No interface for %s!\n", qzdebugstr_guid(riid));
353 return E_NOINTERFACE;
356 ULONG WINAPI PullPin_Release(IPin *iface)
358 PullPin *This = (PullPin *)iface;
359 ULONG refCount = InterlockedDecrement(&This->pin.refCount);
361 TRACE("(%p)->() Release from %d\n", This, refCount + 1);
365 WaitForSingleObject(This->hEventStateChanged, INFINITE);
366 assert(!This->hThread);
369 IMemAllocator_Release(This->pAlloc);
371 IAsyncReader_Release(This->pReader);
372 CloseHandle(This->thread_sleepy);
373 CloseHandle(This->hEventStateChanged);
374 This->thread_lock.DebugInfo->Spare[0] = 0;
375 DeleteCriticalSection(&This->thread_lock);
382 static void PullPin_Flush(PullPin *This)
384 IMediaSample *pSample;
385 TRACE("Flushing!\n");
389 /* Flush outstanding samples */
390 IAsyncReader_BeginFlush(This->pReader);
396 IAsyncReader_WaitForNext(This->pReader, 0, &pSample, &dwUser);
401 assert(!IMediaSample_GetActualDataLength(pSample));
403 IMediaSample_Release(pSample);
406 IAsyncReader_EndFlush(This->pReader);
410 static void PullPin_Thread_Process(PullPin *This)
413 IMediaSample * pSample = NULL;
414 ALLOCATOR_PROPERTIES allocProps;
416 hr = IMemAllocator_GetProperties(This->pAlloc, &allocProps);
418 This->cbAlign = allocProps.cbAlign;
420 if (This->rtCurrent < This->rtStart)
421 This->rtCurrent = MEDIATIME_FROM_BYTES(ALIGNDOWN(BYTES_FROM_MEDIATIME(This->rtStart), This->cbAlign));
425 if (This->rtCurrent >= This->rtStop)
427 IPin_EndOfStream((IPin *)This);
431 /* There is no sample in our buffer */
432 hr = This->fnCustomRequest(This->pUserData);
435 ERR("Request error: %x\n", hr);
437 EnterCriticalSection(This->pin.pCritSec);
438 SetEvent(This->hEventStateChanged);
439 LeaveCriticalSection(This->pin.pCritSec);
446 TRACE("Process sample\n");
449 hr = IAsyncReader_WaitForNext(This->pReader, 10000, &pSample, &dwUser);
451 /* Return an empty sample on error to the implementation in case it does custom parsing, so it knows it's gone */
454 hr = This->fnSampleProc(This->pUserData, pSample, dwUser);
458 /* FIXME: This is not well handled yet! */
459 ERR("Processing error: %x\n", hr);
460 if (hr == VFW_E_TIMEOUT)
470 IMediaSample_Release(pSample);
473 } while (This->rtCurrent < This->rtStop && hr == S_OK && !This->stop_playback);
475 /* Sample was rejected, and we are asked to terminate */
478 IMediaSample_Release(pSample);
481 /* Can't reset state to Sleepy here because that might race, instead PauseProcessing will do that for us
482 * Flush remaining samples
485 This->fnDone(This->pUserData);
487 TRACE("End: %08x, %d\n", hr, This->stop_playback);
490 static void PullPin_Thread_Pause(PullPin *This)
494 EnterCriticalSection(This->pin.pCritSec);
495 This->state = Req_Sleepy;
496 SetEvent(This->hEventStateChanged);
497 LeaveCriticalSection(This->pin.pCritSec);
500 static void PullPin_Thread_Stop(PullPin *This)
502 TRACE("(%p)->()\n", This);
504 EnterCriticalSection(This->pin.pCritSec);
506 CloseHandle(This->hThread);
507 This->hThread = NULL;
508 SetEvent(This->hEventStateChanged);
510 LeaveCriticalSection(This->pin.pCritSec);
512 IBaseFilter_Release(This->pin.pinInfo.pFilter);
518 static DWORD WINAPI PullPin_Thread_Main(LPVOID pv)
521 CoInitializeEx(NULL, COINIT_MULTITHREADED);
527 WaitForSingleObject(This->thread_sleepy, INFINITE);
529 TRACE("State: %d\n", This->state);
533 case Req_Die: PullPin_Thread_Stop(This); break;
534 case Req_Run: PullPin_Thread_Process(This); break;
535 case Req_Pause: PullPin_Thread_Pause(This); break;
536 case Req_Sleepy: ERR("Should not be signalled with SLEEPY!\n"); break;
537 default: ERR("Unknown state request: %d\n", This->state); break;
543 static HRESULT PullPin_InitProcessing(PullPin * This)
547 TRACE("(%p)->()\n", This);
549 /* if we are connected */
554 WaitForSingleObject(This->hEventStateChanged, INFINITE);
555 EnterCriticalSection(This->pin.pCritSec);
557 assert(!This->hThread);
558 assert(This->state == Req_Die);
559 assert(This->stop_playback);
560 assert(WaitForSingleObject(This->thread_sleepy, 0) == WAIT_TIMEOUT);
561 This->state = Req_Sleepy;
563 /* AddRef the filter to make sure it and it's pins will be around
564 * as long as the thread */
565 IBaseFilter_AddRef(This->pin.pinInfo.pFilter);
568 This->hThread = CreateThread(NULL, 0, PullPin_Thread_Main, This, 0, &dwThreadId);
571 hr = HRESULT_FROM_WIN32(GetLastError());
572 IBaseFilter_Release(This->pin.pinInfo.pFilter);
577 SetEvent(This->hEventStateChanged);
578 /* If assert fails, that means a command was not processed before the thread previously terminated */
580 LeaveCriticalSection(This->pin.pCritSec);
583 TRACE(" -- %x\n", hr);
588 HRESULT PullPin_StartProcessing(PullPin * This)
590 /* if we are connected */
591 TRACE("(%p)->()\n", This);
594 assert(This->hThread);
596 PullPin_WaitForStateChange(This, INFINITE);
598 assert(This->state == Req_Sleepy);
601 assert(WaitForSingleObject(This->thread_sleepy, 0) == WAIT_TIMEOUT);
602 This->state = Req_Run;
603 This->stop_playback = 0;
604 ResetEvent(This->hEventStateChanged);
605 SetEvent(This->thread_sleepy);
611 HRESULT PullPin_PauseProcessing(PullPin * This)
613 /* if we are connected */
614 TRACE("(%p)->()\n", This);
617 assert(This->hThread);
619 PullPin_WaitForStateChange(This, INFINITE);
621 EnterCriticalSection(This->pin.pCritSec);
623 assert(!This->stop_playback);
624 assert(This->state == Req_Run|| This->state == Req_Sleepy);
626 assert(WaitForSingleObject(This->thread_sleepy, 0) == WAIT_TIMEOUT);
627 This->state = Req_Pause;
628 This->stop_playback = 1;
629 ResetEvent(This->hEventStateChanged);
630 SetEvent(This->thread_sleepy);
632 LeaveCriticalSection(This->pin.pCritSec);
638 static HRESULT PullPin_StopProcessing(PullPin * This)
640 TRACE("(%p)->()\n", This);
642 /* if we are alive */
643 assert(This->hThread);
645 PullPin_WaitForStateChange(This, INFINITE);
647 assert(This->state == Req_Pause || This->state == Req_Sleepy);
649 This->stop_playback = 1;
650 This->state = Req_Die;
651 assert(WaitForSingleObject(This->thread_sleepy, 0) == WAIT_TIMEOUT);
652 ResetEvent(This->hEventStateChanged);
653 SetEvent(This->thread_sleepy);
657 HRESULT PullPin_WaitForStateChange(PullPin * This, DWORD dwMilliseconds)
659 if (WaitForSingleObject(This->hEventStateChanged, dwMilliseconds) == WAIT_TIMEOUT)
664 HRESULT WINAPI PullPin_QueryAccept(IPin * iface, const AM_MEDIA_TYPE * pmt)
666 PullPin *This = (PullPin *)iface;
668 TRACE("(%p/%p)->(%p)\n", This, iface, pmt);
670 return (This->fnQueryAccept(This->pUserData, pmt) == S_OK ? S_OK : S_FALSE);
673 HRESULT WINAPI PullPin_EndOfStream(IPin * iface)
675 FIXME("(%p)->() stub\n", iface);
677 return SendFurther( iface, deliver_endofstream, NULL, NULL );
680 HRESULT WINAPI PullPin_BeginFlush(IPin * iface)
682 PullPin *This = (PullPin *)iface;
683 TRACE("(%p)->()\n", This);
685 EnterCriticalSection(This->pin.pCritSec);
687 SendFurther( iface, deliver_beginflush, NULL, NULL );
689 LeaveCriticalSection(This->pin.pCritSec);
691 EnterCriticalSection(&This->thread_lock);
694 IAsyncReader_BeginFlush(This->pReader);
695 PullPin_WaitForStateChange(This, INFINITE);
697 if (This->hThread && This->state == Req_Run)
699 PullPin_PauseProcessing(This);
700 PullPin_WaitForStateChange(This, INFINITE);
703 LeaveCriticalSection(&This->thread_lock);
705 EnterCriticalSection(This->pin.pCritSec);
707 This->fnCleanProc(This->pUserData);
709 LeaveCriticalSection(This->pin.pCritSec);
714 HRESULT WINAPI PullPin_EndFlush(IPin * iface)
716 PullPin *This = (PullPin *)iface;
718 TRACE("(%p)->()\n", iface);
720 /* Send further first: Else a race condition might terminate processing early */
721 EnterCriticalSection(This->pin.pCritSec);
722 SendFurther( iface, deliver_endflush, NULL, NULL );
723 LeaveCriticalSection(This->pin.pCritSec);
725 EnterCriticalSection(&This->thread_lock);
730 IAsyncReader_EndFlush(This->pReader);
732 IBaseFilter_GetState(This->pin.pinInfo.pFilter, INFINITE, &state);
734 if (state != State_Stopped)
735 PullPin_StartProcessing(This);
737 PullPin_WaitForStateChange(This, INFINITE);
739 LeaveCriticalSection(&This->thread_lock);
744 HRESULT WINAPI PullPin_Disconnect(IPin *iface)
747 PullPin *This = (PullPin *)iface;
751 EnterCriticalSection(This->pin.pCritSec);
753 if (FAILED(hr = IMemAllocator_Decommit(This->pAlloc)))
754 ERR("Allocator decommit failed with error %x. Possible memory leak\n", hr);
756 if (This->pin.pConnectedTo)
758 IPin_Release(This->pin.pConnectedTo);
759 This->pin.pConnectedTo = NULL;
760 PullPin_StopProcessing(This);
762 FreeMediaType(&This->pin.mtCurrent);
763 ZeroMemory(&This->pin.mtCurrent, sizeof(This->pin.mtCurrent));
769 LeaveCriticalSection(This->pin.pCritSec);
774 HRESULT WINAPI PullPin_NewSegment(IPin * iface, REFERENCE_TIME tStart, REFERENCE_TIME tStop, double dRate)
777 FIXME("(%p)->(%s, %s, %g) stub\n", iface, wine_dbgstr_longlong(tStart), wine_dbgstr_longlong(tStop), dRate);
779 args.tStart = tStart;
783 return SendFurther( iface, deliver_newsegment, &args, NULL );
786 static const IPinVtbl PullPin_Vtbl =
788 PullPin_QueryInterface,
791 BaseInputPinImpl_Connect,
792 PullPin_ReceiveConnection,
794 BasePinImpl_ConnectedTo,
795 BasePinImpl_ConnectionMediaType,
796 BasePinImpl_QueryPinInfo,
797 BasePinImpl_QueryDirection,
800 BasePinImpl_EnumMediaTypes,
801 BasePinImpl_QueryInternalConnections,