dlls/winegstreamer/gstdemux.c

   1 /*
   2  * GStreamer splitter + decoder, adapted from parser.c
   3  *
   4  * Copyright 2010 Maarten Lankhorst for CodeWeavers
   5  * Copyright 2010 Aric Stewart for CodeWeavers
   6  *
   7  * This library is free software; you can redistribute it and/or
   8  * modify it under the terms of the GNU Lesser General Public
   9  * License as published by the Free Software Foundation; either
  10  * version 2.1 of the License, or (at your option) any later version.
  11  *
  12  * This library is distributed in the hope that it will be useful,
  13  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * Lesser General Public License for more details.
  16  *
  17  * You should have received a copy of the GNU Lesser General Public
  18  * License along with this library; if not, write to the Free Software
  19  * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA
  20  */
  21
  22 #include "config.h"
  23
  24 #include <gst/gst.h>
  25 #include <gst/video/video.h>
  26 #include <gst/audio/audio.h>
  27
  28 #include "gst_private.h"
  29 #include "gst_guids.h"
  30 #include "gst_cbs.h"
  31
  32 #include "vfwmsgs.h"
  33 #include "amvideo.h"
  34
  35 #include "wine/unicode.h"
  36 #include "wine/debug.h"
  37
  38 #include <assert.h>
  39
  40 #include "dvdmedia.h"
  41 #include "mmreg.h"
  42 #include "ks.h"
  43 #include "initguid.h"
  44 #include "ksmedia.h"
  45
  46 WINE_DEFAULT_DEBUG_CHANNEL(gstreamer);
  47
  48 static pthread_key_t wine_gst_key;
  49
  50 typedef struct GSTOutPin GSTOutPin;
  51 typedef struct GSTInPin {
  52     BasePin pin;
  53     IAsyncReader *pReader;
  54     IMemAllocator *pAlloc;
  55 } GSTInPin;
  56
  57 typedef struct GSTImpl {
  58     BaseFilter filter;
  59
  60     GSTInPin pInputPin;
  61     GSTOutPin **ppPins;
  62     LONG cStreams;
  63
  64     LONGLONG filesize;
  65
  66     BOOL discont, initial, ignore_flush;
  67     GstElement *gstfilter;
  68     GstPad *my_src, *their_sink;
  69     GstBus *bus;
  70     guint64 start, nextofs, nextpullofs, stop;
  71     ALLOCATOR_PROPERTIES props;
  72     HANDLE event, changed_ofs;
  73
  74     HANDLE push_thread;
  75 } GSTImpl;
  76
  77 struct GSTOutPin {
  78     BaseOutputPin pin;
  79     IQualityControl IQualityControl_iface;
  80
  81     GstPad *their_src;
  82     GstPad *my_sink;
  83     GstBufferPool *gstpool;
  84     int isaud, isvid;
  85     AM_MEDIA_TYPE * pmt;
  86     HANDLE caps_event;
  87     GstSegment *segment;
  88     SourceSeeking seek;
  89 };
  90
  91 const char* media_quark_string = "media-sample";
  92
  93 static const WCHAR wcsInputPinName[] = {'i','n','p','u','t',' ','p','i','n',0};
  94 static const IMediaSeekingVtbl GST_Seeking_Vtbl;
  95 static const IPinVtbl GST_OutputPin_Vtbl;
  96 static const IPinVtbl GST_InputPin_Vtbl;
  97 static const IBaseFilterVtbl GST_Vtbl;
  98 static const IQualityControlVtbl GSTOutPin_QualityControl_Vtbl;
  99
 100 static HRESULT GST_AddPin(GSTImpl *This, const PIN_INFO *piOutput, const AM_MEDIA_TYPE *amt);
 101 static HRESULT GST_RemoveOutputPins(GSTImpl *This);
 102 static HRESULT WINAPI GST_ChangeCurrent(IMediaSeeking *iface);
 103 static HRESULT WINAPI GST_ChangeStop(IMediaSeeking *iface);
 104 static HRESULT WINAPI GST_ChangeRate(IMediaSeeking *iface);
 105
 106 void mark_wine_thread(void)
 107 {
 108     /* set it to non-NULL to indicate that this is a Wine thread */
 109     pthread_setspecific(wine_gst_key, &wine_gst_key);
 110 }
 111
 112 BOOL is_wine_thread(void)
 113 {
 114     return pthread_getspecific(wine_gst_key) != NULL;
 115 }
 116
 117 static gboolean amt_from_gst_caps_audio(GstCaps *caps, AM_MEDIA_TYPE *amt)
 118 {
 119     WAVEFORMATEXTENSIBLE *wfe;
 120     WAVEFORMATEX *wfx;
 121     gint32 depth = 0, bpp = 0;
 122     GstAudioInfo ainfo;
 123
 124     if (!gst_audio_info_from_caps (&ainfo, caps))
 125         return FALSE;
 126
 127     wfe = CoTaskMemAlloc(sizeof(*wfe));
 128     wfx = (WAVEFORMATEX*)wfe;
 129     amt->majortype = MEDIATYPE_Audio;
 130     amt->subtype = MEDIASUBTYPE_PCM;
 131     amt->formattype = FORMAT_WaveFormatEx;
 132     amt->pbFormat = (BYTE*)wfe;
 133     amt->cbFormat = sizeof(*wfe);
 134     amt->bFixedSizeSamples = 0;
 135     amt->bTemporalCompression = 1;
 136     amt->lSampleSize = 0;
 137     amt->pUnk = NULL;
 138
 139     wfx->wFormatTag = WAVE_FORMAT_EXTENSIBLE;
 140
 141     wfx->nChannels = ainfo.channels;
 142     wfx->nSamplesPerSec = ainfo.rate;
 143     depth = GST_AUDIO_INFO_WIDTH(&ainfo);
 144     bpp = GST_AUDIO_INFO_DEPTH(&ainfo);
 145
 146     if (!depth || depth > 32 || depth % 8)
 147         depth = bpp;
 148     else if (!bpp)
 149         bpp = depth;
 150     wfe->Samples.wValidBitsPerSample = depth;
 151     wfx->wBitsPerSample = bpp;
 152     wfx->cbSize = sizeof(*wfe)-sizeof(*wfx);
 153     switch (wfx->nChannels) {
 154         case 1: wfe->dwChannelMask = KSAUDIO_SPEAKER_MONO; break;
 155         case 2: wfe->dwChannelMask = KSAUDIO_SPEAKER_STEREO; break;
 156         case 4: wfe->dwChannelMask = KSAUDIO_SPEAKER_SURROUND; break;
 157         case 5: wfe->dwChannelMask = (KSAUDIO_SPEAKER_5POINT1 & ~SPEAKER_LOW_FREQUENCY); break;
 158         case 6: wfe->dwChannelMask = KSAUDIO_SPEAKER_5POINT1; break;
 159         case 8: wfe->dwChannelMask = KSAUDIO_SPEAKER_7POINT1; break;
 160         default:
 161         wfe->dwChannelMask = 0;
 162     }
 163     if (GST_AUDIO_INFO_IS_FLOAT(&ainfo)) {
 164         wfe->SubFormat = KSDATAFORMAT_SUBTYPE_IEEE_FLOAT;
 165         wfx->wBitsPerSample = wfe->Samples.wValidBitsPerSample = 32;
 166     } else {
 167         wfe->SubFormat = KSDATAFORMAT_SUBTYPE_PCM;
 168         if (wfx->nChannels <= 2 && bpp <= 16 && depth == bpp)  {
 169             wfx->wFormatTag = WAVE_FORMAT_PCM;
 170             wfx->cbSize = 0;
 171         }
 172     }
 173     wfx->nBlockAlign = wfx->nChannels * wfx->wBitsPerSample/8;
 174     wfx->nAvgBytesPerSec = wfx->nSamplesPerSec * wfx->nBlockAlign;
 175     return TRUE;
 176 }
 177
 178 static gboolean amt_from_gst_caps_video(GstCaps *caps, AM_MEDIA_TYPE *amt)
 179 {
 180     VIDEOINFOHEADER *vih = CoTaskMemAlloc(sizeof(*vih));
 181     BITMAPINFOHEADER *bih = &vih->bmiHeader;
 182     gint32 width = 0, height = 0, nom = 0, denom = 0;
 183     GstVideoInfo vinfo;
 184
 185     if (!gst_video_info_from_caps (&vinfo, caps))
 186         return FALSE;
 187     width = vinfo.width;
 188     height = vinfo.height;
 189     nom = vinfo.fps_n;
 190     denom = vinfo.fps_d;
 191
 192     amt->formattype = FORMAT_VideoInfo;
 193     amt->pbFormat = (BYTE*)vih;
 194     amt->cbFormat = sizeof(*vih);
 195     amt->bFixedSizeSamples = amt->bTemporalCompression = 1;
 196     amt->lSampleSize = 0;
 197     amt->pUnk = NULL;
 198     ZeroMemory(vih, sizeof(*vih));
 199     amt->majortype = MEDIATYPE_Video;
 200     if (GST_VIDEO_INFO_IS_RGB(&vinfo)) {
 201         bih->biBitCount = GST_VIDEO_FORMAT_INFO_BITS(vinfo.finfo);
 202         switch (bih->biBitCount) {
 203             case 16: amt->subtype = MEDIASUBTYPE_RGB555; break;
 204             case 24: amt->subtype = MEDIASUBTYPE_RGB24; break;
 205             case 32: amt->subtype = MEDIASUBTYPE_RGB32; break;
 206             default:
 207                 FIXME("Unknown bpp %u\n", bih->biBitCount);
 208                 return FALSE;
 209         }
 210         bih->biCompression = BI_RGB;
 211     } else {
 212         amt->subtype = MEDIATYPE_Video;
 213         if (!(amt->subtype.Data1 = gst_video_format_to_fourcc(vinfo.finfo->format)))
 214             return FALSE;
 215         switch (amt->subtype.Data1) {
 216             case mmioFOURCC('I','4','2','0'):
 217             case mmioFOURCC('Y','V','1','2'):
 218             case mmioFOURCC('N','V','1','2'):
 219             case mmioFOURCC('N','V','2','1'):
 220                 bih->biBitCount = 12; break;
 221             case mmioFOURCC('Y','U','Y','2'):
 222             case mmioFOURCC('Y','V','Y','U'):
 223                 bih->biBitCount = 16; break;
 224         }
 225         bih->biCompression = amt->subtype.Data1;
 226     }
 227     bih->biSizeImage = width * height * bih->biBitCount / 8;
 228     vih->AvgTimePerFrame = 10000000;
 229     vih->AvgTimePerFrame *= denom;
 230     vih->AvgTimePerFrame /= nom;
 231     vih->rcSource.left = 0;
 232     vih->rcSource.right = width;
 233     vih->rcSource.top = height;
 234     vih->rcSource.bottom = 0;
 235     vih->rcTarget = vih->rcSource;
 236     bih->biSize = sizeof(*bih);
 237     bih->biWidth = width;
 238     bih->biHeight = height;
 239     bih->biPlanes = 1;
 240     return TRUE;
 241 }
 242
 243 static gboolean accept_caps_sink(GstPad *pad, GstCaps *caps)
 244 {
 245     GSTOutPin *pin = gst_pad_get_element_private(pad);
 246     AM_MEDIA_TYPE amt;
 247     GstStructure *arg;
 248     const char *typename;
 249     gboolean ret;
 250
 251     TRACE("%p %p\n", pad, caps);
 252
 253     arg = gst_caps_get_structure(caps, 0);
 254     typename = gst_structure_get_name(arg);
 255     if (!strcmp(typename, "audio/x-raw")) {
 256         if (!pin->isaud) {
 257             ERR("Setting audio caps on non-audio pad?\n");
 258             return FALSE;
 259         }
 260         ret = amt_from_gst_caps_audio(caps, &amt);
 261         FreeMediaType(&amt);
 262         TRACE("+%i\n", ret);
 263         return ret;
 264     } else if (!strcmp(typename, "video/x-raw")) {
 265         if (!pin->isvid) {
 266             ERR("Setting video caps on non-video pad?\n");
 267             return FALSE;
 268         }
 269         ret = amt_from_gst_caps_video(caps, &amt);
 270         FreeMediaType(&amt);
 271         TRACE("-%i\n", ret);
 272         return ret;
 273     } else {
 274         FIXME("Unhandled type \"%s\"\n", typename);
 275         return FALSE;
 276     }
 277 }
 278
 279 static gboolean setcaps_sink(GstPad *pad, GstCaps *caps)
 280 {
 281     GSTOutPin *pin = gst_pad_get_element_private(pad);
 282     GSTImpl *This = (GSTImpl *)pin->pin.pin.pinInfo.pFilter;
 283     AM_MEDIA_TYPE amt;
 284     GstStructure *arg;
 285     const char *typename;
 286     gboolean ret;
 287
 288     TRACE("%p %p\n", pad, caps);
 289
 290     arg = gst_caps_get_structure(caps, 0);
 291     typename = gst_structure_get_name(arg);
 292     if (!strcmp(typename, "audio/x-raw")) {
 293         if (!pin->isaud) {
 294             ERR("Setting audio caps on non-audio pad?\n");
 295             return FALSE;
 296         }
 297         ret = amt_from_gst_caps_audio(caps, &amt);
 298     } else if (!strcmp(typename, "video/x-raw")) {
 299         if (!pin->isvid) {
 300             ERR("Setting video caps on non-video pad?\n");
 301             return FALSE;
 302         }
 303         ret = amt_from_gst_caps_video(caps, &amt);
 304         if (ret)
 305             This->props.cbBuffer = max(This->props.cbBuffer, ((VIDEOINFOHEADER*)amt.pbFormat)->bmiHeader.biSizeImage);
 306     } else {
 307         FIXME("Unhandled type \"%s\"\n", typename);
 308         return FALSE;
 309     }
 310     TRACE("Linking returned %i for %s\n", ret, typename);
 311     if (!ret)
 312         return FALSE;
 313     FreeMediaType(pin->pmt);
 314     *pin->pmt = amt;
 315     return TRUE;
 316 }
 317
 318 static gboolean query_sink(GstPad *pad, GstObject *parent, GstQuery *query)
 319 {
 320     switch (GST_QUERY_TYPE (query)) {
 321         case GST_QUERY_ACCEPT_CAPS:
 322         {
 323             GstCaps *caps;
 324             gboolean res;
 325             gst_query_parse_accept_caps(query, &caps);
 326             res = accept_caps_sink(pad, caps);
 327             gst_query_set_accept_caps_result(query, res);
 328             return TRUE; /* FIXME */
 329         }
 330         default:
 331             return gst_pad_query_default (pad, parent, query);
 332     }
 333 }
 334
 335 static gboolean gst_base_src_perform_seek(GSTImpl *This, GstEvent *event)
 336 {
 337     gboolean res = TRUE;
 338     gdouble rate;
 339     GstFormat seek_format;
 340     GstSeekFlags flags;
 341     GstSeekType cur_type, stop_type;
 342     gint64 cur, stop;
 343     gboolean flush;
 344     guint32 seqnum;
 345     GstEvent *tevent;
 346     BOOL thread = !!This->push_thread;
 347
 348     TRACE("%p %p\n", This, event);
 349
 350     gst_event_parse_seek(event, &rate, &seek_format, &flags,
 351                          &cur_type, &cur, &stop_type, &stop);
 352
 353     if (seek_format != GST_FORMAT_BYTES) {
 354         FIXME("Not handling other format %i\n", seek_format);
 355         return FALSE;
 356     }
 357
 358     flush = flags & GST_SEEK_FLAG_FLUSH;
 359     seqnum = gst_event_get_seqnum(event);
 360
 361     /* send flush start */
 362     if (flush) {
 363         tevent = gst_event_new_flush_start();
 364         gst_event_set_seqnum(tevent, seqnum);
 365         gst_pad_push_event(This->my_src, tevent);
 366         if (This->pInputPin.pReader)
 367             IAsyncReader_BeginFlush(This->pInputPin.pReader);
 368         if (thread)
 369             gst_pad_set_active(This->my_src, 1);
 370     }
 371
 372     This->nextofs = This->start = cur;
 373
 374     /* and prepare to continue streaming */
 375     if (flush) {
 376         tevent = gst_event_new_flush_stop(TRUE);
 377         gst_event_set_seqnum(tevent, seqnum);
 378         gst_pad_push_event(This->my_src, tevent);
 379         if (This->pInputPin.pReader)
 380             IAsyncReader_EndFlush(This->pInputPin.pReader);
 381         if (thread)
 382             gst_pad_set_active(This->my_src, 1);
 383     }
 384
 385     return res;
 386 }
 387
 388 static gboolean event_src(GstPad *pad, GstObject *parent, GstEvent *event)
 389 {
 390     GSTImpl *This = gst_pad_get_element_private(pad);
 391
 392     TRACE("%p %p\n", pad, event);
 393
 394     switch (event->type) {
 395         case GST_EVENT_SEEK:
 396             return gst_base_src_perform_seek(This, event);
 397         case GST_EVENT_FLUSH_START:
 398             EnterCriticalSection(&This->filter.csFilter);
 399             if (This->pInputPin.pReader)
 400                 IAsyncReader_BeginFlush(This->pInputPin.pReader);
 401             LeaveCriticalSection(&This->filter.csFilter);
 402             break;
 403         case GST_EVENT_FLUSH_STOP:
 404             EnterCriticalSection(&This->filter.csFilter);
 405             if (This->pInputPin.pReader)
 406                 IAsyncReader_EndFlush(This->pInputPin.pReader);
 407             LeaveCriticalSection(&This->filter.csFilter);
 408             break;
 409         default:
 410             FIXME("%p (%u) stub\n", event, event->type);
 411         case GST_EVENT_TAG:
 412         case GST_EVENT_QOS:
 413             return gst_pad_event_default(pad, parent, event);
 414     }
 415     return TRUE;
 416 }
 417
 418 static gboolean event_sink(GstPad *pad, GstObject *parent, GstEvent *event)
 419 {
 420     GSTOutPin *pin = gst_pad_get_element_private(pad);
 421
 422     TRACE("%p %p\n", pad, event);
 423
 424     switch (event->type) {
 425         case GST_EVENT_SEGMENT: {
 426             gdouble rate, applied_rate;
 427             gint64 stop, pos;
 428             const GstSegment *segment;
 429
 430             gst_event_parse_segment(event, &segment);
 431
 432             pos = segment->position;
 433             stop = segment->stop;
 434             rate = segment->rate;
 435             applied_rate = segment->applied_rate;
 436
 437             if (segment->format != GST_FORMAT_TIME) {
 438                 FIXME("Ignoring new segment because of format %i\n", segment->format);
 439                 return TRUE;
 440             }
 441
 442             gst_segment_copy_into(segment, pin->segment);
 443
 444             pos /= 100;
 445
 446             if (stop > 0)
 447                 stop /= 100;
 448
 449             if (pin->pin.pin.pConnectedTo)
 450                 IPin_NewSegment(pin->pin.pin.pConnectedTo, pos, stop, rate*applied_rate);
 451
 452             return TRUE;
 453         }
 454         case GST_EVENT_EOS:
 455             if (pin->pin.pin.pConnectedTo)
 456                 IPin_EndOfStream(pin->pin.pin.pConnectedTo);
 457             return TRUE;
 458         case GST_EVENT_FLUSH_START:
 459             if (((GSTImpl *)pin->pin.pin.pinInfo.pFilter)->ignore_flush) {
 460                 /* gst-plugins-base prior to 1.7 contains a bug which causes
 461                  * our sink pins to receive a flush-start event when the
 462                  * decodebin changes from PAUSED to READY (including
 463                  * PLAYING->PAUSED->READY), but no matching flush-stop event is
 464                  * sent. See <gst-plugins-base.git:60bad4815db966a8e4). Here we
 465                  * unset the flushing flag to avoid the problem. */
 466                 TRACE("Working around gst <1.7 bug, ignoring FLUSH_START\n");
 467                 GST_PAD_UNSET_FLUSHING (pad);
 468                 return TRUE;
 469             }
 470             if (pin->pin.pin.pConnectedTo)
 471                 IPin_BeginFlush(pin->pin.pin.pConnectedTo);
 472             return TRUE;
 473         case GST_EVENT_FLUSH_STOP:
 474             gst_segment_init(pin->segment, GST_FORMAT_TIME);
 475             if (pin->pin.pin.pConnectedTo)
 476                 IPin_EndFlush(pin->pin.pin.pConnectedTo);
 477             return TRUE;
 478         case GST_EVENT_CAPS: {
 479             GstCaps *caps;
 480             gst_event_parse_caps(event, &caps);
 481             return setcaps_sink(pad, caps);
 482         }
 483         default:
 484             TRACE("%p stub %s\n", event, gst_event_type_get_name(event->type));
 485             return gst_pad_event_default(pad, parent, event);
 486     }
 487 }
 488
 489 static void release_sample(void *data)
 490 {
 491     ULONG ret;
 492     ret = IMediaSample_Release((IMediaSample *)data);
 493     TRACE("Releasing %p returns %u\n", data, ret);
 494 }
 495
 496 static DWORD CALLBACK push_data(LPVOID iface)
 497 {
 498     LONGLONG maxlen, curlen;
 499     GSTImpl *This = iface;
 500     IMediaSample *buf;
 501     DWORD_PTR user;
 502     HRESULT hr;
 503
 504     if (!This->stop)
 505         IAsyncReader_Length(This->pInputPin.pReader, &maxlen, &curlen);
 506     else
 507         maxlen = This->stop;
 508
 509     TRACE("Waiting..\n");
 510
 511     WaitForSingleObject(This->event, INFINITE);
 512
 513     TRACE("Starting..\n");
 514     for (;;) {
 515         REFERENCE_TIME tStart, tStop;
 516         ULONG len;
 517         GstBuffer *gstbuf;
 518         gsize bufsize;
 519         BYTE *data;
 520         int ret;
 521
 522         TRACE("pAlloc: %p\n", This->pInputPin.pAlloc);
 523         hr = IMemAllocator_GetBuffer(This->pInputPin.pAlloc, &buf, NULL, NULL, 0);
 524         if (FAILED(hr))
 525             break;
 526
 527         if (This->nextofs >= maxlen)
 528             break;
 529         len = IMediaSample_GetSize(buf);
 530         if (This->nextofs + len > maxlen)
 531             len = maxlen - This->nextofs;
 532
 533         tStart = MEDIATIME_FROM_BYTES(This->nextofs);
 534         tStop = tStart + MEDIATIME_FROM_BYTES(len);
 535         IMediaSample_SetTime(buf, &tStart, &tStop);
 536
 537         hr = IAsyncReader_Request(This->pInputPin.pReader, buf, 0);
 538         if (FAILED(hr)) {
 539             IMediaSample_Release(buf);
 540             break;
 541         }
 542         This->nextofs += len;
 543         hr = IAsyncReader_WaitForNext(This->pInputPin.pReader, -1, &buf, &user);
 544         if (FAILED(hr) || !buf) {
 545             if (buf)
 546                 IMediaSample_Release(buf);
 547             break;
 548         }
 549
 550         IMediaSample_GetPointer(buf, &data);
 551         bufsize = IMediaSample_GetActualDataLength(buf);
 552         gstbuf = gst_buffer_new_wrapped_full(0, data, bufsize, 0, bufsize, buf, release_sample_wrapper);
 553         IMediaSample_AddRef(buf);
 554         gst_mini_object_set_qdata(GST_MINI_OBJECT(gstbuf), g_quark_from_static_string(media_quark_string), buf, release_sample_wrapper);
 555         if (!gstbuf) {
 556             IMediaSample_Release(buf);
 557             break;
 558         }
 559         gstbuf->duration = gstbuf->pts = -1;
 560         ret = gst_pad_push(This->my_src, gstbuf);
 561         if (ret >= 0)
 562             hr = S_OK;
 563         else
 564             ERR("Sending returned: %i\n", ret);
 565         if (ret == GST_FLOW_ERROR)
 566             hr = E_FAIL;
 567         else if (ret == GST_FLOW_FLUSHING)
 568             hr = VFW_E_WRONG_STATE;
 569         if (hr != S_OK)
 570             break;
 571     }
 572
 573     gst_pad_push_event(This->my_src, gst_event_new_eos());
 574
 575     TRACE("Almost stopping.. %08x\n", hr);
 576     do {
 577         IAsyncReader_WaitForNext(This->pInputPin.pReader, 0, &buf, &user);
 578         if (buf)
 579             IMediaSample_Release(buf);
 580     } while (buf);
 581
 582     TRACE("Stopping.. %08x\n", hr);
 583     return 0;
 584 }
 585
 586 static HRESULT WINAPI GST_OutPin_QueryAccept(IPin *iface, const AM_MEDIA_TYPE *pmt)
 587 {
 588     GSTOutPin *pin = (GSTOutPin*)iface;
 589     FIXME("stub %p\n", pin);
 590     return S_OK;
 591 }
 592
 593 static GstFlowReturn got_data_sink(GstPad *pad, GstObject *parent, GstBuffer *buf)
 594 {
 595     GSTOutPin *pin = gst_pad_get_element_private(pad);
 596     GSTImpl *This = (GSTImpl *)pin->pin.pin.pinInfo.pFilter;
 597     HRESULT hr;
 598     BYTE *ptr = NULL;
 599     IMediaSample *sample;
 600     GstMapInfo info;
 601
 602     TRACE("%p %p\n", pad, buf);
 603
 604     if (This->initial) {
 605         gst_buffer_unref(buf);
 606         TRACE("Triggering %p %p\n", pad, pin->caps_event);
 607         SetEvent(pin->caps_event);
 608         return GST_FLOW_OK;
 609     }
 610
 611     hr = BaseOutputPinImpl_GetDeliveryBuffer(&pin->pin, &sample, NULL, NULL, 0);
 612
 613     if (hr == VFW_E_NOT_CONNECTED) {
 614         gst_buffer_unref(buf);
 615         return GST_FLOW_NOT_LINKED;
 616     }
 617
 618     if (FAILED(hr)) {
 619         gst_buffer_unref(buf);
 620         ERR("Could not get a delivery buffer (%x), returning GST_FLOW_FLUSHING\n", hr);
 621         return GST_FLOW_FLUSHING;
 622     }
 623
 624     gst_buffer_map(buf, &info, GST_MAP_READ);
 625
 626     hr = IMediaSample_SetActualDataLength(sample, info.size);
 627     if(FAILED(hr)){
 628         WARN("SetActualDataLength failed: %08x\n", hr);
 629         return GST_FLOW_FLUSHING;
 630     }
 631
 632     IMediaSample_GetPointer(sample, &ptr);
 633
 634     memcpy(ptr, info.data, info.size);
 635
 636     gst_buffer_unmap(buf, &info);
 637
 638     if (GST_BUFFER_PTS_IS_VALID(buf)) {
 639         REFERENCE_TIME rtStart = gst_segment_to_running_time(pin->segment, GST_FORMAT_TIME, buf->pts);
 640         if (rtStart >= 0)
 641             rtStart /= 100;
 642
 643         if (GST_BUFFER_DURATION_IS_VALID(buf)) {
 644             REFERENCE_TIME tStart = buf->pts / 100;
 645             REFERENCE_TIME tStop = (buf->pts + buf->duration) / 100;
 646             REFERENCE_TIME rtStop;
 647             rtStop = gst_segment_to_running_time(pin->segment, GST_FORMAT_TIME, buf->pts + buf->duration);
 648             if (rtStop >= 0)
 649                 rtStop /= 100;
 650             TRACE("Current time on %p: %i to %i ms\n", pin, (int)(rtStart / 10000), (int)(rtStop / 10000));
 651             IMediaSample_SetTime(sample, &rtStart, rtStop >= 0 ? &rtStop : NULL);
 652             IMediaSample_SetMediaTime(sample, &tStart, &tStop);
 653         } else {
 654             IMediaSample_SetTime(sample, rtStart >= 0 ? &rtStart : NULL, NULL);
 655             IMediaSample_SetMediaTime(sample, NULL, NULL);
 656         }
 657     } else {
 658         IMediaSample_SetTime(sample, NULL, NULL);
 659         IMediaSample_SetMediaTime(sample, NULL, NULL);
 660     }
 661
 662     IMediaSample_SetDiscontinuity(sample, GST_BUFFER_FLAG_IS_SET(buf, GST_BUFFER_FLAG_DISCONT));
 663     IMediaSample_SetPreroll(sample, GST_BUFFER_FLAG_IS_SET(buf, GST_BUFFER_FLAG_LIVE));
 664     IMediaSample_SetSyncPoint(sample, !GST_BUFFER_FLAG_IS_SET(buf, GST_BUFFER_FLAG_DELTA_UNIT));
 665
 666     if (!pin->pin.pin.pConnectedTo)
 667         hr = VFW_E_NOT_CONNECTED;
 668     else
 669         hr = IMemInputPin_Receive(pin->pin.pMemInputPin, sample);
 670
 671     TRACE("sending sample returned: %08x\n", hr);
 672
 673     gst_buffer_unref(buf);
 674     IMediaSample_Release(sample);
 675
 676     if (hr == VFW_E_NOT_CONNECTED)
 677         return GST_FLOW_NOT_LINKED;
 678
 679     if (FAILED(hr))
 680         return GST_FLOW_FLUSHING;
 681
 682     return GST_FLOW_OK;
 683 }
 684
 685 static GstFlowReturn request_buffer_src(GstPad *pad, GstObject *parent, guint64 ofs, guint len, GstBuffer **buf)
 686 {
 687     GSTImpl *This = gst_pad_get_element_private(pad);
 688     HRESULT hr;
 689     GstMapInfo info;
 690
 691     TRACE("%p %s %i %p\n", pad, wine_dbgstr_longlong(ofs), len, buf);
 692
 693     *buf = NULL;
 694     if (ofs == (guint64)-1)
 695         ofs = This->nextpullofs;
 696     if (ofs >= This->filesize) {
 697         WARN("Reading past eof: %s, %u\n", wine_dbgstr_longlong(ofs), len);
 698         return GST_FLOW_EOS;
 699     }
 700     if (len + ofs > This->filesize)
 701         len = This->filesize - ofs;
 702     This->nextpullofs = ofs + len;
 703
 704     *buf = gst_buffer_new_and_alloc(len);
 705     gst_buffer_map(*buf, &info, GST_MAP_WRITE);
 706     hr = IAsyncReader_SyncRead(This->pInputPin.pReader, ofs, len, info.data);
 707     gst_buffer_unmap(*buf, &info);
 708     if (FAILED(hr)) {
 709         ERR("Returned %08x\n", hr);
 710         return GST_FLOW_ERROR;
 711     }
 712     return GST_FLOW_OK;
 713 }
 714
 715 static DWORD CALLBACK push_data_init(LPVOID iface)
 716 {
 717     GSTImpl *This = iface;
 718     DWORD64 ofs = 0;
 719
 720     TRACE("Starting..\n");
 721     for (;;) {
 722         GstBuffer *buf;
 723         GstFlowReturn ret = request_buffer_src(This->my_src, NULL, ofs, 4096, &buf);
 724         if (ret < 0) {
 725             ERR("Obtaining buffer returned: %i\n", ret);
 726             break;
 727         }
 728         ret = gst_pad_push(This->my_src, buf);
 729         ofs += 4096;
 730         if (ret)
 731             TRACE("Sending returned: %i\n", ret);
 732         if (ret < 0)
 733             break;
 734     }
 735     TRACE("Stopping..\n");
 736     return 0;
 737 }
 738
 739 static void removed_decoded_pad(GstElement *bin, GstPad *pad, gpointer user)
 740 {
 741     GSTImpl *This = (GSTImpl*)user;
 742     int x;
 743     GSTOutPin *pin;
 744
 745     TRACE("%p %p %p\n", This, bin, pad);
 746
 747     EnterCriticalSection(&This->filter.csFilter);
 748     for (x = 0; x < This->cStreams; ++x) {
 749         if (This->ppPins[x]->their_src == pad)
 750             break;
 751     }
 752     if (x == This->cStreams)
 753         goto out;
 754     pin = This->ppPins[x];
 755     gst_pad_unlink(pin->their_src, pin->my_sink);
 756     gst_object_unref(pin->their_src);
 757     pin->their_src = NULL;
 758 out:
 759     TRACE("Removed %i/%i\n", x, This->cStreams);
 760     LeaveCriticalSection(&This->filter.csFilter);
 761 }
 762
 763 static void init_new_decoded_pad(GstElement *bin, GstPad *pad, GSTImpl *This)
 764 {
 765     HRESULT hr;
 766     PIN_INFO piOutput;
 767     const char *typename;
 768     char *name;
 769     AM_MEDIA_TYPE amt = {{0}};
 770     GstCaps *caps;
 771     GstStructure *arg;
 772     GstPad *mypad;
 773     GSTOutPin *pin;
 774     int ret;
 775     int isvid = 0, isaud = 0;
 776     gchar my_name[1024];
 777
 778     TRACE("%p %p %p\n", This, bin, pad);
 779
 780     piOutput.dir = PINDIR_OUTPUT;
 781     piOutput.pFilter = (IBaseFilter *)This;
 782     name = gst_pad_get_name(pad);
 783     MultiByteToWideChar(CP_UNIXCP, 0, name, -1, piOutput.achName, sizeof(piOutput.achName) / sizeof(piOutput.achName[0]) - 1);
 784     TRACE("Name: %s\n", name);
 785     strcpy(my_name, "qz_sink_");
 786     strcat(my_name, name);
 787     g_free(name);
 788     piOutput.achName[sizeof(piOutput.achName) / sizeof(piOutput.achName[0]) - 1] = 0;
 789
 790     caps = gst_pad_query_caps(pad, NULL);
 791     caps = gst_caps_make_writable(caps);
 792     arg = gst_caps_get_structure(caps, 0);
 793     typename = gst_structure_get_name(arg);
 794
 795     mypad = gst_pad_new(my_name, GST_PAD_SINK);
 796     gst_pad_set_chain_function(mypad, got_data_sink_wrapper);
 797     gst_pad_set_event_function(mypad, event_sink_wrapper);
 798     gst_pad_set_query_function(mypad, query_sink_wrapper);
 799
 800     if (!strcmp(typename, "audio/x-raw")) {
 801         isaud = 1;
 802     } else if (!strcmp(typename, "video/x-raw")) {
 803         isvid = 1;
 804     } else {
 805         FIXME("Unknown type \'%s\'\n", typename);
 806         return;
 807     }
 808
 809     hr = GST_AddPin(This, &piOutput, &amt);
 810     if (FAILED(hr)) {
 811         ERR("%08x\n", hr);
 812         return;
 813     }
 814
 815     pin = This->ppPins[This->cStreams - 1];
 816     gst_pad_set_element_private(mypad, pin);
 817     pin->my_sink = mypad;
 818     pin->isaud = isaud;
 819     pin->isvid = isvid;
 820
 821     gst_segment_init(pin->segment, GST_FORMAT_TIME);
 822     ret = gst_pad_link(pad, mypad);
 823
 824     gst_pad_set_active(mypad, 1);
 825
 826     TRACE("Linking: %i\n", ret);
 827
 828     if (ret >= 0) {
 829         pin->their_src = pad;
 830         gst_object_ref(pin->their_src);
 831     }
 832 }
 833
 834 static void existing_new_pad(GstElement *bin, GstPad *pad, gpointer user)
 835 {
 836     GSTImpl *This = (GSTImpl*)user;
 837     int x;
 838
 839     TRACE("%p %p %p\n", This, bin, pad);
 840
 841     if (gst_pad_is_linked(pad))
 842         return;
 843
 844     /* Still holding our own lock */
 845     if (This->initial) {
 846         init_new_decoded_pad(bin, pad, This);
 847         return;
 848     }
 849
 850     EnterCriticalSection(&This->filter.csFilter);
 851     for (x = 0; x < This->cStreams; ++x) {
 852         GSTOutPin *pin = This->ppPins[x];
 853         if (!pin->their_src) {
 854             gst_segment_init(pin->segment, GST_FORMAT_TIME);
 855             if (gst_pad_link(pad, pin->my_sink) >= 0) {
 856                 pin->their_src = pad;
 857                 gst_object_ref(pin->their_src);
 858                 TRACE("Relinked\n");
 859                 LeaveCriticalSection(&This->filter.csFilter);
 860                 return;
 861             }
 862         }
 863     }
 864     init_new_decoded_pad(bin, pad, This);
 865     LeaveCriticalSection(&This->filter.csFilter);
 866 }
 867
 868 static gboolean query_function(GstPad *pad, GstObject *parent, GstQuery *query)
 869 {
 870     GSTImpl *This = gst_pad_get_element_private(pad);
 871     GstFormat format;
 872     int ret;
 873     LONGLONG duration;
 874
 875     TRACE("%p %p %p\n", This, pad, query);
 876
 877     switch (GST_QUERY_TYPE(query)) {
 878         case GST_QUERY_DURATION:
 879             gst_query_parse_duration (query, &format, NULL);
 880             if (format == GST_FORMAT_PERCENT) {
 881                 gst_query_set_duration (query, GST_FORMAT_PERCENT, GST_FORMAT_PERCENT_MAX);
 882                 return TRUE;
 883             }
 884             ret = gst_pad_query_convert (pad, GST_FORMAT_BYTES, This->filesize, format, &duration);
 885             gst_query_set_duration(query, format, duration);
 886             return ret;
 887         case GST_QUERY_SEEKING:
 888             gst_query_parse_seeking (query, &format, NULL, NULL, NULL);
 889             TRACE("Seeking %i %i\n", format, GST_FORMAT_BYTES);
 890             if (format != GST_FORMAT_BYTES)
 891                 return FALSE;
 892             gst_query_set_seeking(query, GST_FORMAT_BYTES, 1, 0, This->filesize);
 893             return TRUE;
 894         case GST_QUERY_SCHEDULING:
 895             gst_query_set_scheduling(query, GST_SCHEDULING_FLAG_SEEKABLE, 1, -1, 0);
 896             gst_query_add_scheduling_mode(query, GST_PAD_MODE_PUSH);
 897             gst_query_add_scheduling_mode(query, GST_PAD_MODE_PULL);
 898             return TRUE;
 899         default:
 900             TRACE("Unhandled query type: %s\n", GST_QUERY_TYPE_NAME(query));
 901             return FALSE;
 902     }
 903 }
 904
 905 static gboolean activate_push(GstPad *pad, gboolean activate)
 906 {
 907     GSTImpl *This = gst_pad_get_element_private(pad);
 908
 909     TRACE("%p %p %u\n", This, pad, activate);
 910
 911     EnterCriticalSection(&This->filter.csFilter);
 912     if (!activate) {
 913         TRACE("Deactivating\n");
 914         if (!This->initial)
 915             IAsyncReader_BeginFlush(This->pInputPin.pReader);
 916         if (This->push_thread) {
 917             WaitForSingleObject(This->push_thread, -1);
 918             CloseHandle(This->push_thread);
 919             This->push_thread = NULL;
 920         }
 921         if (!This->initial)
 922             IAsyncReader_EndFlush(This->pInputPin.pReader);
 923         if (This->filter.state == State_Stopped)
 924             This->nextofs = This->start;
 925     } else if (!This->push_thread) {
 926         TRACE("Activating\n");
 927         if (This->initial)
 928             This->push_thread = CreateThread(NULL, 0, push_data_init, This, 0, NULL);
 929         else
 930             This->push_thread = CreateThread(NULL, 0, push_data, This, 0, NULL);
 931     }
 932     LeaveCriticalSection(&This->filter.csFilter);
 933     return TRUE;
 934 }
 935
 936 static gboolean activate_mode(GstPad *pad, GstObject *parent, GstPadMode mode, gboolean activate)
 937 {
 938     TRACE("%p %p 0x%x %u\n", pad, parent, mode, activate);
 939     switch (mode) {
 940       case GST_PAD_MODE_PULL:
 941         return TRUE;
 942       case GST_PAD_MODE_PUSH:
 943         return activate_push(pad, activate);
 944       default:
 945         return FALSE;
 946     }
 947     return FALSE;
 948 }
 949
 950 static void no_more_pads(GstElement *decodebin, gpointer user)
 951 {
 952     GSTImpl *This = (GSTImpl*)user;
 953     TRACE("%p %p\n", This, decodebin);
 954     SetEvent(This->event);
 955 }
 956
 957 static GstAutoplugSelectResult autoplug_blacklist(GstElement *bin, GstPad *pad, GstCaps *caps, GstElementFactory *fact, gpointer user)
 958 {
 959     const char *name = gst_element_factory_get_longname(fact);
 960
 961     if (strstr(name, "Player protection")) {
 962         WARN("Blacklisted a/52 decoder because it only works in Totem\n");
 963         return GST_AUTOPLUG_SELECT_SKIP;
 964     }
 965     if (!strcmp(name, "Fluendo Hardware Accelerated Video Decoder")) {
 966         WARN("Disabled video acceleration since it breaks in wine\n");
 967         return GST_AUTOPLUG_SELECT_SKIP;
 968     }
 969     TRACE("using \"%s\"\n", name);
 970     return GST_AUTOPLUG_SELECT_TRY;
 971 }
 972
 973 static GstBusSyncReply watch_bus(GstBus *bus, GstMessage *msg, gpointer data)
 974 {
 975     GSTImpl *This = data;
 976     GError *err = NULL;
 977     gchar *dbg_info = NULL;
 978
 979     TRACE("%p %p %p\n", This, bus, msg);
 980
 981     if (GST_MESSAGE_TYPE(msg) & GST_MESSAGE_ERROR) {
 982         gst_message_parse_error(msg, &err, &dbg_info);
 983         FIXME("%s: %s\n", GST_OBJECT_NAME(msg->src), err->message);
 984         WARN("%s\n", dbg_info);
 985         SetEvent(This->event);
 986     } else if (GST_MESSAGE_TYPE(msg) & GST_MESSAGE_WARNING) {
 987         gst_message_parse_warning(msg, &err, &dbg_info);
 988         WARN("%s: %s\n", GST_OBJECT_NAME(msg->src), err->message);
 989         WARN("%s\n", dbg_info);
 990     }
 991     if (err)
 992         g_error_free(err);
 993     g_free(dbg_info);
 994     return GST_BUS_DROP;
 995 }
 996
 997 static void unknown_type(GstElement *bin, GstPad *pad, GstCaps *caps, gpointer user)
 998 {
 999     gchar *strcaps = gst_caps_to_string(caps);
1000     FIXME("Could not find a filter for caps: %s\n", strcaps);
1001     g_free(strcaps);
1002 }
1003
1004 static HRESULT GST_Connect(GSTInPin *pPin, IPin *pConnectPin, ALLOCATOR_PROPERTIES *props)
1005 {
1006     GSTImpl *This = (GSTImpl*)pPin->pin.pinInfo.pFilter;
1007     HRESULT hr;
1008     int ret, i;
1009     LONGLONG avail, duration;
1010     GstStaticPadTemplate src_template = GST_STATIC_PAD_TEMPLATE(
1011         "quartz_src",
1012         GST_PAD_SRC,
1013         GST_PAD_ALWAYS,
1014         GST_STATIC_CAPS_ANY);
1015
1016     TRACE("%p %p %p\n", pPin, pConnectPin, props);
1017
1018     This->props = *props;
1019     IAsyncReader_Length(pPin->pReader, &This->filesize, &avail);
1020
1021     if (!This->bus) {
1022         This->bus = gst_bus_new();
1023         gst_bus_set_sync_handler(This->bus, watch_bus_wrapper, This, NULL);
1024     }
1025
1026     This->gstfilter = gst_element_factory_make("decodebin", NULL);
1027     if (!This->gstfilter) {
1028         FIXME("Could not make source filter, are gstreamer-plugins-* installed for %u bits?\n",
1029               8 * (int)sizeof(void*));
1030         return E_FAIL;
1031     }
1032     gst_element_set_bus(This->gstfilter, This->bus);
1033     g_signal_connect(This->gstfilter, "pad-added", G_CALLBACK(existing_new_pad_wrapper), This);
1034     g_signal_connect(This->gstfilter, "pad-removed", G_CALLBACK(removed_decoded_pad_wrapper), This);
1035     g_signal_connect(This->gstfilter, "autoplug-select", G_CALLBACK(autoplug_blacklist_wrapper), This);
1036     g_signal_connect(This->gstfilter, "unknown-type", G_CALLBACK(unknown_type_wrapper), This);
1037
1038     This->my_src = gst_pad_new_from_static_template(&src_template, "quartz-src");
1039     gst_pad_set_getrange_function(This->my_src, request_buffer_src_wrapper);
1040     gst_pad_set_query_function(This->my_src, query_function_wrapper);
1041     gst_pad_set_activatemode_function(This->my_src, activate_mode_wrapper);
1042     gst_pad_set_event_function(This->my_src, event_src_wrapper);
1043     gst_pad_set_element_private (This->my_src, This);
1044     This->their_sink = gst_element_get_static_pad(This->gstfilter, "sink");
1045
1046     g_signal_connect(This->gstfilter, "no-more-pads", G_CALLBACK(no_more_pads_wrapper), This);
1047     ret = gst_pad_link(This->my_src, This->their_sink);
1048     if (ret < 0) {
1049         ERR("Returns: %i\n", ret);
1050         return E_FAIL;
1051     }
1052     This->start = This->nextofs = This->nextpullofs = This->stop = 0;
1053
1054     /* Add initial pins */
1055     This->initial = This->discont = TRUE;
1056     ResetEvent(This->event);
1057     gst_element_set_state(This->gstfilter, GST_STATE_PLAYING);
1058     WaitForSingleObject(This->event, -1);
1059     gst_element_get_state(This->gstfilter, NULL, NULL, -1);
1060
1061     if (ret < 0) {
1062         WARN("Ret: %i\n", ret);
1063         hr = E_FAIL;
1064     } else if (!This->cStreams) {
1065         FIXME("GStreamer could not find any streams\n");
1066         hr = E_FAIL;
1067     } else {
1068         gst_pad_query_duration(This->ppPins[0]->their_src, GST_FORMAT_TIME, &duration);
1069         for (i = 0; i < This->cStreams; ++i) {
1070             This->ppPins[i]->seek.llDuration = This->ppPins[i]->seek.llStop = duration / 100;
1071             This->ppPins[i]->seek.llCurrent = 0;
1072             if (!This->ppPins[i]->seek.llDuration)
1073                 This->ppPins[i]->seek.dwCapabilities = 0;
1074             WaitForSingleObject(This->ppPins[i]->caps_event, -1);
1075         }
1076         hr = S_OK;
1077     }
1078     *props = This->props;
1079
1080     This->ignore_flush = TRUE;
1081     gst_element_set_state(This->gstfilter, GST_STATE_READY);
1082     gst_element_get_state(This->gstfilter, NULL, NULL, -1);
1083     This->ignore_flush = FALSE;
1084
1085     This->initial = FALSE;
1086
1087     /* don't set active during test-play, as we don't want to push/pull data
1088      * from the source yet */
1089     gst_pad_set_active(This->my_src, 1);
1090
1091     This->nextofs = This->nextpullofs = 0;
1092     return hr;
1093 }
1094
1095 static inline GSTOutPin *impl_from_IMediaSeeking( IMediaSeeking *iface )
1096 {
1097     return CONTAINING_RECORD(iface, GSTOutPin, seek.IMediaSeeking_iface);
1098 }
1099
1100 static IPin* WINAPI GST_GetPin(BaseFilter *iface, int pos)
1101 {
1102     GSTImpl *This = (GSTImpl *)iface;
1103     TRACE("%p: Asking for pos %x\n", This, pos);
1104
1105     if (pos > This->cStreams || pos < 0)
1106         return NULL;
1107     if (!pos)
1108     {
1109         IPin_AddRef((IPin*)&This->pInputPin);
1110         return (IPin*)&This->pInputPin;
1111     }
1112     else
1113     {
1114         IPin_AddRef((IPin*)This->ppPins[pos - 1]);
1115         return (IPin*)This->ppPins[pos - 1];
1116     }
1117 }
1118
1119 static LONG WINAPI GST_GetPinCount(BaseFilter *iface)
1120 {
1121     GSTImpl *This = (GSTImpl *)iface;
1122     TRACE("%p -> %u\n", This, This->cStreams + 1);
1123     return (This->cStreams + 1);
1124 }
1125
1126 static const BaseFilterFuncTable BaseFuncTable = {
1127     GST_GetPin,
1128     GST_GetPinCount
1129 };
1130
1131 IUnknown * CALLBACK Gstreamer_Splitter_create(IUnknown *pUnkOuter, HRESULT *phr)
1132 {
1133     IUnknown *obj = NULL;
1134     PIN_INFO *piInput;
1135     GSTImpl *This;
1136
1137     TRACE("%p %p\n", pUnkOuter, phr);
1138
1139     if (!Gstreamer_init())
1140     {
1141         *phr = E_FAIL;
1142         return NULL;
1143     }
1144
1145     mark_wine_thread();
1146
1147     This = CoTaskMemAlloc(sizeof(*This));
1148     obj = (IUnknown*)This;
1149     if (!This)
1150     {
1151         *phr = E_OUTOFMEMORY;
1152         return NULL;
1153     }
1154
1155     BaseFilter_Init(&This->filter, &GST_Vtbl, &CLSID_Gstreamer_Splitter, (DWORD_PTR)(__FILE__ ": GSTImpl.csFilter"), &BaseFuncTable);
1156
1157     This->cStreams = 0;
1158     This->ppPins = NULL;
1159     This->push_thread = NULL;
1160     This->event = CreateEventW(NULL, 0, 0, NULL);
1161     This->bus = NULL;
1162
1163     piInput = &This->pInputPin.pin.pinInfo;
1164     piInput->dir = PINDIR_INPUT;
1165     piInput->pFilter = (IBaseFilter *)This;
1166     lstrcpynW(piInput->achName, wcsInputPinName, sizeof(piInput->achName) / sizeof(piInput->achName[0]));
1167     This->pInputPin.pin.IPin_iface.lpVtbl = &GST_InputPin_Vtbl;
1168     This->pInputPin.pin.refCount = 1;
1169     This->pInputPin.pin.pConnectedTo = NULL;
1170     This->pInputPin.pin.pCritSec = &This->filter.csFilter;
1171     ZeroMemory(&This->pInputPin.pin.mtCurrent, sizeof(AM_MEDIA_TYPE));
1172     *phr = S_OK;
1173
1174     TRACE("returning %p\n", obj);
1175
1176     return obj;
1177 }
1178
1179 static void GST_Destroy(GSTImpl *This)
1180 {
1181     IPin *connected = NULL;
1182     ULONG pinref;
1183     HRESULT hr;
1184
1185     TRACE("Destroying %p\n", This);
1186
1187     CloseHandle(This->event);
1188
1189     /* Don't need to clean up output pins, disconnecting input pin will do that */
1190     IPin_ConnectedTo((IPin *)&This->pInputPin, &connected);
1191     if (connected) {
1192         hr = IPin_Disconnect(connected);
1193         assert(hr == S_OK);
1194         IPin_Release(connected);
1195         hr = IPin_Disconnect((IPin *)&This->pInputPin);
1196         assert(hr == S_OK);
1197     }
1198     pinref = IPin_Release((IPin *)&This->pInputPin);
1199     if (pinref) {
1200         /* Valgrind could find this, if I kill it here */
1201         ERR("pinref should be null, is %u, destroying anyway\n", pinref);
1202         assert((LONG)pinref > 0);
1203
1204         while (pinref)
1205             pinref = IPin_Release((IPin *)&This->pInputPin);
1206     }
1207     if (This->bus) {
1208         gst_bus_set_sync_handler(This->bus, NULL, NULL, NULL);
1209         gst_object_unref(This->bus);
1210     }
1211     BaseFilter_Destroy(&This->filter);
1212     CoTaskMemFree(This);
1213 }
1214
1215 static HRESULT WINAPI GST_QueryInterface(IBaseFilter *iface, REFIID riid, LPVOID *ppv)
1216 {
1217     GSTImpl *This = (GSTImpl *)iface;
1218
1219     TRACE("(%p)->(%s, %p)\n", This, debugstr_guid(riid), ppv);
1220
1221     *ppv = NULL;
1222
1223     if (IsEqualIID(riid, &IID_IUnknown))
1224         *ppv = This;
1225     else if (IsEqualIID(riid, &IID_IPersist))
1226         *ppv = This;
1227     else if (IsEqualIID(riid, &IID_IMediaFilter))
1228         *ppv = This;
1229     else if (IsEqualIID(riid, &IID_IBaseFilter))
1230         *ppv = This;
1231
1232     if (*ppv) {
1233         IUnknown_AddRef((IUnknown *)(*ppv));
1234         return S_OK;
1235     }
1236
1237     if (!IsEqualIID(riid, &IID_IPin) && !IsEqualIID(riid, &IID_IVideoWindow) &&
1238         !IsEqualIID(riid, &IID_IAMFilterMiscFlags))
1239         FIXME("No interface for %s!\n", debugstr_guid(riid));
1240
1241     return E_NOINTERFACE;
1242 }
1243
1244 static ULONG WINAPI GST_Release(IBaseFilter *iface)
1245 {
1246     GSTImpl *This = (GSTImpl *)iface;
1247     ULONG refCount = InterlockedDecrement(&This->filter.refCount);
1248
1249     TRACE("(%p)->() Release from %d\n", This, refCount + 1);
1250
1251     if (!refCount)
1252         GST_Destroy(This);
1253
1254     return refCount;
1255 }
1256
1257 static HRESULT WINAPI GST_Stop(IBaseFilter *iface)
1258 {
1259     GSTImpl *This = (GSTImpl *)iface;
1260
1261     TRACE("(%p)\n", This);
1262
1263     mark_wine_thread();
1264
1265     if (This->gstfilter) {
1266         This->ignore_flush = TRUE;
1267         gst_element_set_state(This->gstfilter, GST_STATE_READY);
1268         gst_element_get_state(This->gstfilter, NULL, NULL, -1);
1269         This->ignore_flush = FALSE;
1270     }
1271     return S_OK;
1272 }
1273
1274 static HRESULT WINAPI GST_Pause(IBaseFilter *iface)
1275 {
1276     HRESULT hr = S_OK;
1277     GSTImpl *This = (GSTImpl *)iface;
1278     GstState now;
1279     GstStateChangeReturn ret;
1280
1281     TRACE("(%p)\n", This);
1282
1283     if (!This->gstfilter)
1284         return VFW_E_NOT_CONNECTED;
1285
1286     mark_wine_thread();
1287
1288     gst_element_get_state(This->gstfilter, &now, NULL, -1);
1289     if (now == GST_STATE_PAUSED)
1290         return S_OK;
1291     if (now != GST_STATE_PLAYING)
1292         hr = IBaseFilter_Run(iface, -1);
1293     if (FAILED(hr))
1294         return hr;
1295     ret = gst_element_set_state(This->gstfilter, GST_STATE_PAUSED);
1296     if (ret == GST_STATE_CHANGE_ASYNC)
1297         hr = S_FALSE;
1298     return hr;
1299 }
1300
1301 static HRESULT WINAPI GST_Run(IBaseFilter *iface, REFERENCE_TIME tStart)
1302 {
1303     HRESULT hr = S_OK;
1304     GSTImpl *This = (GSTImpl *)iface;
1305     ULONG i;
1306     GstState now;
1307     HRESULT hr_any = VFW_E_NOT_CONNECTED;
1308
1309     TRACE("(%p)->(%s)\n", This, wine_dbgstr_longlong(tStart));
1310
1311     mark_wine_thread();
1312
1313     if (!This->gstfilter)
1314         return VFW_E_NOT_CONNECTED;
1315
1316     EnterCriticalSection(&This->filter.csFilter);
1317     This->filter.rtStreamStart = tStart;
1318     LeaveCriticalSection(&This->filter.csFilter);
1319
1320     gst_element_get_state(This->gstfilter, &now, NULL, -1);
1321     if (now == GST_STATE_PLAYING)
1322         return S_OK;
1323     if (now == GST_STATE_PAUSED) {
1324         GstStateChangeReturn ret;
1325         ret = gst_element_set_state(This->gstfilter, GST_STATE_PLAYING);
1326         if (ret == GST_STATE_CHANGE_ASYNC)
1327             return S_FALSE;
1328         return S_OK;
1329     }
1330
1331     EnterCriticalSection(&This->filter.csFilter);
1332     gst_element_set_state(This->gstfilter, GST_STATE_PLAYING);
1333     This->filter.rtStreamStart = tStart;
1334
1335     for (i = 0; i < This->cStreams; i++) {
1336         hr = BaseOutputPinImpl_Active((BaseOutputPin *)This->ppPins[i]);
1337         if (SUCCEEDED(hr)) {
1338             hr_any = hr;
1339         }
1340     }
1341     hr = hr_any;
1342     LeaveCriticalSection(&This->filter.csFilter);
1343
1344     return hr;
1345 }
1346
1347 static HRESULT WINAPI GST_GetState(IBaseFilter *iface, DWORD dwMilliSecsTimeout, FILTER_STATE *pState)
1348 {
1349     GSTImpl *This = (GSTImpl *)iface;
1350     HRESULT hr = S_OK;
1351     GstState now, pending;
1352     GstStateChangeReturn ret;
1353
1354     TRACE("(%p)->(%d, %p)\n", This, dwMilliSecsTimeout, pState);
1355
1356     mark_wine_thread();
1357
1358     if (!This->gstfilter) {
1359         *pState = State_Stopped;
1360         return S_OK;
1361     }
1362
1363     ret = gst_element_get_state(This->gstfilter, &now, &pending, dwMilliSecsTimeout == INFINITE ? -1 : dwMilliSecsTimeout * 1000);
1364
1365     if (ret == GST_STATE_CHANGE_ASYNC)
1366         hr = VFW_S_STATE_INTERMEDIATE;
1367     else
1368         pending = now;
1369
1370     switch (pending) {
1371         case GST_STATE_PAUSED: *pState = State_Paused; return hr;
1372         case GST_STATE_PLAYING: *pState = State_Running; return hr;
1373         default: *pState = State_Stopped; return hr;
1374     }
1375 }
1376
1377 static HRESULT WINAPI GST_FindPin(IBaseFilter *iface, LPCWSTR Id, IPin **ppPin)
1378 {
1379     FIXME("(%p)->(%s,%p) stub\n", iface, debugstr_w(Id), ppPin);
1380     return E_NOTIMPL;
1381 }
1382
1383 static const IBaseFilterVtbl GST_Vtbl = {
1384     GST_QueryInterface,
1385     BaseFilterImpl_AddRef,
1386     GST_Release,
1387     BaseFilterImpl_GetClassID,
1388     GST_Stop,
1389     GST_Pause,
1390     GST_Run,
1391     GST_GetState,
1392     BaseFilterImpl_SetSyncSource,
1393     BaseFilterImpl_GetSyncSource,
1394     BaseFilterImpl_EnumPins,
1395     GST_FindPin,
1396     BaseFilterImpl_QueryFilterInfo,
1397     BaseFilterImpl_JoinFilterGraph,
1398     BaseFilterImpl_QueryVendorInfo
1399 };
1400
1401 static HRESULT WINAPI GST_ChangeCurrent(IMediaSeeking *iface)
1402 {
1403     GSTOutPin *This = impl_from_IMediaSeeking(iface);
1404     TRACE("(%p)\n", This);
1405     return S_OK;
1406 }
1407
1408 static HRESULT WINAPI GST_ChangeStop(IMediaSeeking *iface)
1409 {
1410     GSTOutPin *This = impl_from_IMediaSeeking(iface);
1411     TRACE("(%p)\n", This);
1412     return S_OK;
1413 }
1414
1415 static HRESULT WINAPI GST_ChangeRate(IMediaSeeking *iface)
1416 {
1417     GSTOutPin *This = impl_from_IMediaSeeking(iface);
1418     GstEvent *ev = gst_event_new_seek(This->seek.dRate, GST_FORMAT_TIME, 0, GST_SEEK_TYPE_NONE, -1, GST_SEEK_TYPE_NONE, -1);
1419     TRACE("(%p) New rate %g\n", This, This->seek.dRate);
1420     mark_wine_thread();
1421     gst_pad_push_event(This->my_sink, ev);
1422     return S_OK;
1423 }
1424
1425 static HRESULT WINAPI GST_Seeking_QueryInterface(IMediaSeeking *iface, REFIID riid, void **ppv)
1426 {
1427     GSTOutPin *This = impl_from_IMediaSeeking(iface);
1428     return IUnknown_QueryInterface((IUnknown *)This, riid, ppv);
1429 }
1430
1431 static ULONG WINAPI GST_Seeking_AddRef(IMediaSeeking *iface)
1432 {
1433     GSTOutPin *This = impl_from_IMediaSeeking(iface);
1434     return IUnknown_AddRef((IUnknown *)This);
1435 }
1436
1437 static ULONG WINAPI GST_Seeking_Release(IMediaSeeking *iface)
1438 {
1439     GSTOutPin *This = impl_from_IMediaSeeking(iface);
1440     return IUnknown_Release((IUnknown *)This);
1441 }
1442
1443 static HRESULT WINAPI GST_Seeking_GetCurrentPosition(IMediaSeeking *iface, REFERENCE_TIME *pos)
1444 {
1445     GSTOutPin *This = impl_from_IMediaSeeking(iface);
1446
1447     TRACE("(%p)->(%p)\n", This, pos);
1448
1449     if (!pos)
1450         return E_POINTER;
1451
1452     mark_wine_thread();
1453
1454     if (!This->their_src) {
1455         *pos = This->seek.llCurrent;
1456         TRACE("Cached value\n");
1457         if (This->seek.llDuration)
1458             return S_OK;
1459         else
1460             return E_NOTIMPL;
1461     }
1462
1463     if (!gst_pad_query_position(This->their_src, GST_FORMAT_TIME, pos)) {
1464         WARN("Could not query position\n");
1465         return E_NOTIMPL;
1466     }
1467     *pos /= 100;
1468     This->seek.llCurrent = *pos;
1469     return S_OK;
1470 }
1471
1472 static GstSeekType type_from_flags(DWORD flags)
1473 {
1474     switch (flags & AM_SEEKING_PositioningBitsMask) {
1475     case AM_SEEKING_NoPositioning:
1476         return GST_SEEK_TYPE_NONE;
1477     case AM_SEEKING_AbsolutePositioning:
1478     case AM_SEEKING_RelativePositioning:
1479         return GST_SEEK_TYPE_SET;
1480     case AM_SEEKING_IncrementalPositioning:
1481         return GST_SEEK_TYPE_END;
1482     }
1483     return GST_SEEK_TYPE_NONE;
1484 }
1485
1486 static HRESULT WINAPI GST_Seeking_SetPositions(IMediaSeeking *iface,
1487         REFERENCE_TIME *pCur, DWORD curflags, REFERENCE_TIME *pStop,
1488         DWORD stopflags)
1489 {
1490     HRESULT hr;
1491     GSTOutPin *This = impl_from_IMediaSeeking(iface);
1492     GstSeekFlags f = 0;
1493     GstSeekType curtype, stoptype;
1494     GstEvent *e;
1495     gint64 stop_pos = 0, curr_pos = 0;
1496
1497     TRACE("(%p)->(%p, 0x%x, %p, 0x%x)\n", This, pCur, curflags, pStop, stopflags);
1498
1499     mark_wine_thread();
1500
1501     if (!This->seek.llDuration)
1502         return E_NOTIMPL;
1503
1504     hr = SourceSeekingImpl_SetPositions(iface, pCur, curflags, pStop, stopflags);
1505     if (!This->their_src)
1506         return hr;
1507
1508     curtype = type_from_flags(curflags);
1509     stoptype = type_from_flags(stopflags);
1510     if (curflags & AM_SEEKING_SeekToKeyFrame)
1511         f |= GST_SEEK_FLAG_KEY_UNIT;
1512     if (curflags & AM_SEEKING_Segment)
1513         f |= GST_SEEK_FLAG_SEGMENT;
1514     if (!(curflags & AM_SEEKING_NoFlush))
1515         f |= GST_SEEK_FLAG_FLUSH;
1516
1517     if (((curflags & AM_SEEKING_PositioningBitsMask) == AM_SEEKING_RelativePositioning) ||
1518         ((stopflags & AM_SEEKING_PositioningBitsMask) == AM_SEEKING_RelativePositioning)) {
1519         gint64 tmp_pos;
1520         gst_pad_query_position (This->my_sink, GST_FORMAT_TIME, &tmp_pos);
1521         if ((curflags & AM_SEEKING_PositioningBitsMask) == AM_SEEKING_RelativePositioning)
1522             curr_pos = tmp_pos;
1523         if ((stopflags & AM_SEEKING_PositioningBitsMask) == AM_SEEKING_RelativePositioning)
1524             stop_pos = tmp_pos;
1525     }
1526
1527     e = gst_event_new_seek(This->seek.dRate, GST_FORMAT_TIME, f, curtype, pCur ? curr_pos + *pCur * 100 : -1, stoptype, pStop ? stop_pos + *pStop * 100 : -1);
1528     if (gst_pad_push_event(This->my_sink, e))
1529         return S_OK;
1530     else
1531         return E_NOTIMPL;
1532 }
1533
1534 static const IMediaSeekingVtbl GST_Seeking_Vtbl =
1535 {
1536     GST_Seeking_QueryInterface,
1537     GST_Seeking_AddRef,
1538     GST_Seeking_Release,
1539     SourceSeekingImpl_GetCapabilities,
1540     SourceSeekingImpl_CheckCapabilities,
1541     SourceSeekingImpl_IsFormatSupported,
1542     SourceSeekingImpl_QueryPreferredFormat,
1543     SourceSeekingImpl_GetTimeFormat,
1544     SourceSeekingImpl_IsUsingTimeFormat,
1545     SourceSeekingImpl_SetTimeFormat,
1546     SourceSeekingImpl_GetDuration,
1547     SourceSeekingImpl_GetStopPosition,
1548     GST_Seeking_GetCurrentPosition,
1549     SourceSeekingImpl_ConvertTimeFormat,
1550     GST_Seeking_SetPositions,
1551     SourceSeekingImpl_GetPositions,
1552     SourceSeekingImpl_GetAvailable,
1553     SourceSeekingImpl_SetRate,
1554     SourceSeekingImpl_GetRate,
1555     SourceSeekingImpl_GetPreroll
1556 };
1557
1558 static inline GSTOutPin *impl_from_IQualityControl( IQualityControl *iface )
1559 {
1560     return (GSTOutPin*)CONTAINING_RECORD(iface, GSTOutPin, IQualityControl_iface);
1561 }
1562
1563 static HRESULT WINAPI GST_QualityControl_QueryInterface(IQualityControl *iface, REFIID riid, void **ppv)
1564 {
1565     GSTOutPin *pin = impl_from_IQualityControl(iface);
1566     return IPin_QueryInterface((IPin*)pin, riid, ppv);
1567 }
1568
1569 static ULONG WINAPI GST_QualityControl_AddRef(IQualityControl *iface)
1570 {
1571     GSTOutPin *pin = impl_from_IQualityControl(iface);
1572     return IPin_AddRef((IPin*)pin);
1573 }
1574
1575 static ULONG WINAPI GST_QualityControl_Release(IQualityControl *iface)
1576 {
1577     GSTOutPin *pin = impl_from_IQualityControl(iface);
1578     return IPin_Release((IPin*)pin);
1579 }
1580
1581 static HRESULT WINAPI GST_QualityControl_Notify(IQualityControl *iface, IBaseFilter *sender, Quality qm)
1582 {
1583     GSTOutPin *pin = impl_from_IQualityControl(iface);
1584     REFERENCE_TIME late = qm.Late;
1585     TRACE("(%p)->(%p, qm)\n", pin, sender);
1586     mark_wine_thread();
1587     if (qm.Late < 0 && -qm.Late > qm.TimeStamp)
1588         late = -qm.TimeStamp;
1589     gst_pad_push_event(pin->my_sink, gst_event_new_qos(late <= 0 ? GST_QOS_TYPE_OVERFLOW : GST_QOS_TYPE_UNDERFLOW, 1000./qm.Proportion, late*100, qm.TimeStamp*100));
1590     return S_OK;
1591 }
1592
1593 static HRESULT WINAPI GST_QualityControl_SetSink(IQualityControl *iface, IQualityControl *tonotify)
1594 {
1595     GSTOutPin *pin = impl_from_IQualityControl(iface);
1596     TRACE("(%p)->(%p)\n", pin, pin);
1597     /* Do nothing */
1598     return S_OK;
1599 }
1600
1601 static const IQualityControlVtbl GSTOutPin_QualityControl_Vtbl = {
1602     GST_QualityControl_QueryInterface,
1603     GST_QualityControl_AddRef,
1604     GST_QualityControl_Release,
1605     GST_QualityControl_Notify,
1606     GST_QualityControl_SetSink
1607 };
1608
1609 static HRESULT WINAPI GSTOutPin_QueryInterface(IPin *iface, REFIID riid, void **ppv)
1610 {
1611     GSTOutPin *This = (GSTOutPin *)iface;
1612
1613     TRACE("(%p)->(%s, %p)\n", This, debugstr_guid(riid), ppv);
1614
1615     *ppv = NULL;
1616
1617     if (IsEqualIID(riid, &IID_IUnknown))
1618         *ppv = iface;
1619     else if (IsEqualIID(riid, &IID_IPin))
1620         *ppv = iface;
1621     else if (IsEqualIID(riid, &IID_IMediaSeeking))
1622         *ppv = &This->seek;
1623     else if (IsEqualIID(riid, &IID_IQualityControl))
1624         *ppv = &This->IQualityControl_iface;
1625
1626     if (*ppv) {
1627         IUnknown_AddRef((IUnknown *)(*ppv));
1628         return S_OK;
1629     }
1630     FIXME("No interface for %s!\n", debugstr_guid(riid));
1631     return E_NOINTERFACE;
1632 }
1633
1634 static ULONG WINAPI GSTOutPin_Release(IPin *iface)
1635 {
1636     GSTOutPin *This = (GSTOutPin *)iface;
1637     ULONG refCount = InterlockedDecrement(&This->pin.pin.refCount);
1638
1639     TRACE("(%p)->() Release from %d\n", This, refCount + 1);
1640
1641     mark_wine_thread();
1642
1643     if (!refCount) {
1644         if (This->their_src) {
1645             gst_pad_unlink(This->their_src, This->my_sink);
1646             gst_object_unref(This->their_src);
1647         }
1648         gst_object_unref(This->my_sink);
1649         CloseHandle(This->caps_event);
1650         DeleteMediaType(This->pmt);
1651         FreeMediaType(&This->pin.pin.mtCurrent);
1652         gst_segment_free(This->segment);
1653         if(This->gstpool)
1654             gst_object_unref(This->gstpool);
1655         if (This->pin.pAllocator)
1656             IMemAllocator_Release(This->pin.pAllocator);
1657         CoTaskMemFree(This);
1658         return 0;
1659     }
1660     return refCount;
1661 }
1662
1663 static HRESULT WINAPI GSTOutPin_GetMediaType(BasePin *iface, int iPosition, AM_MEDIA_TYPE *pmt)
1664 {
1665     GSTOutPin *This = (GSTOutPin *)iface;
1666
1667     TRACE("(%p)->(%i, %p)\n", This, iPosition, pmt);
1668
1669     if (iPosition < 0)
1670         return E_INVALIDARG;
1671
1672     if (iPosition > 0)
1673         return VFW_S_NO_MORE_ITEMS;
1674
1675     CopyMediaType(pmt, This->pmt);
1676
1677     return S_OK;
1678 }
1679
1680 static HRESULT WINAPI GSTOutPin_DecideBufferSize(BaseOutputPin *iface, IMemAllocator *pAlloc, ALLOCATOR_PROPERTIES *ppropInputRequest)
1681 {
1682     GSTOutPin *This = (GSTOutPin *)iface;
1683     TRACE("(%p)->(%p, %p)\n", This, pAlloc, ppropInputRequest);
1684     /* Unused */
1685     return S_OK;
1686 }
1687
1688 static HRESULT WINAPI GSTOutPin_DecideAllocator(BaseOutputPin *iface, IMemInputPin *pPin, IMemAllocator **pAlloc)
1689 {
1690     HRESULT hr;
1691     GSTOutPin *This = (GSTOutPin *)iface;
1692     GSTImpl *GSTfilter = (GSTImpl*)This->pin.pin.pinInfo.pFilter;
1693
1694     TRACE("(%p)->(%p, %p)\n", This, pPin, pAlloc);
1695
1696     *pAlloc = NULL;
1697     if (GSTfilter->pInputPin.pAlloc)
1698     {
1699         hr = IMemInputPin_NotifyAllocator(pPin, GSTfilter->pInputPin.pAlloc, FALSE);
1700         if (SUCCEEDED(hr))
1701         {
1702             *pAlloc = GSTfilter->pInputPin.pAlloc;
1703             IMemAllocator_AddRef(*pAlloc);
1704         }
1705     }
1706     else
1707         hr = VFW_E_NO_ALLOCATOR;
1708
1709     return hr;
1710 }
1711
1712 static HRESULT WINAPI GSTOutPin_BreakConnect(BaseOutputPin *This)
1713 {
1714     HRESULT hr;
1715
1716     TRACE("(%p)->()\n", This);
1717
1718     EnterCriticalSection(This->pin.pCritSec);
1719     if (!This->pin.pConnectedTo || !This->pMemInputPin)
1720         hr = VFW_E_NOT_CONNECTED;
1721     else
1722     {
1723         hr = IPin_Disconnect(This->pin.pConnectedTo);
1724         IPin_Disconnect((IPin *)This);
1725     }
1726     LeaveCriticalSection(This->pin.pCritSec);
1727
1728     return hr;
1729 }
1730
1731 static const IPinVtbl GST_OutputPin_Vtbl = {
1732     GSTOutPin_QueryInterface,
1733     BasePinImpl_AddRef,
1734     GSTOutPin_Release,
1735     BaseOutputPinImpl_Connect,
1736     BaseOutputPinImpl_ReceiveConnection,
1737     BaseOutputPinImpl_Disconnect,
1738     BasePinImpl_ConnectedTo,
1739     BasePinImpl_ConnectionMediaType,
1740     BasePinImpl_QueryPinInfo,
1741     BasePinImpl_QueryDirection,
1742     BasePinImpl_QueryId,
1743     GST_OutPin_QueryAccept,
1744     BasePinImpl_EnumMediaTypes,
1745     BasePinImpl_QueryInternalConnections,
1746     BaseOutputPinImpl_EndOfStream,
1747     BaseOutputPinImpl_BeginFlush,
1748     BaseOutputPinImpl_EndFlush,
1749     BasePinImpl_NewSegment
1750 };
1751
1752 static const BaseOutputPinFuncTable output_BaseOutputFuncTable = {
1753     {
1754         NULL,
1755         BaseOutputPinImpl_AttemptConnection,
1756         BasePinImpl_GetMediaTypeVersion,
1757         GSTOutPin_GetMediaType
1758     },
1759     GSTOutPin_DecideBufferSize,
1760     GSTOutPin_DecideAllocator,
1761     GSTOutPin_BreakConnect
1762 };
1763
1764 static HRESULT GST_AddPin(GSTImpl *This, const PIN_INFO *piOutput, const AM_MEDIA_TYPE *amt)
1765 {
1766     HRESULT hr;
1767     This->ppPins = CoTaskMemRealloc(This->ppPins, (This->cStreams + 1) * sizeof(IPin *));
1768
1769     hr = BaseOutputPin_Construct(&GST_OutputPin_Vtbl, sizeof(GSTOutPin), piOutput, &output_BaseOutputFuncTable, &This->filter.csFilter, (IPin**)(This->ppPins + This->cStreams));
1770     if (SUCCEEDED(hr)) {
1771         GSTOutPin *pin = This->ppPins[This->cStreams];
1772         memset((char*)pin + sizeof(pin->pin), 0, sizeof(GSTOutPin) - sizeof(pin->pin));
1773         pin->pmt = CoTaskMemAlloc(sizeof(AM_MEDIA_TYPE));
1774         CopyMediaType(pin->pmt, amt);
1775         pin->pin.pin.pinInfo.pFilter = (LPVOID)This;
1776         pin->caps_event = CreateEventW(NULL, 0, 0, NULL);
1777         pin->segment = gst_segment_new();
1778         This->cStreams++;
1779         pin->IQualityControl_iface.lpVtbl = &GSTOutPin_QualityControl_Vtbl;
1780         SourceSeeking_Init(&pin->seek, &GST_Seeking_Vtbl, GST_ChangeStop, GST_ChangeCurrent, GST_ChangeRate, &This->filter.csFilter);
1781         BaseFilterImpl_IncrementPinVersion((BaseFilter*)This);
1782     } else
1783         ERR("Failed with error %x\n", hr);
1784     return hr;
1785 }
1786
1787 static HRESULT GST_RemoveOutputPins(GSTImpl *This)
1788 {
1789     HRESULT hr;
1790     ULONG i;
1791     GSTOutPin **ppOldPins = This->ppPins;
1792
1793     TRACE("(%p)\n", This);
1794     mark_wine_thread();
1795
1796     if (!This->gstfilter)
1797         return S_OK;
1798     gst_element_set_bus(This->gstfilter, NULL);
1799     gst_element_set_state(This->gstfilter, GST_STATE_NULL);
1800     gst_pad_unlink(This->my_src, This->their_sink);
1801     gst_object_unref(This->my_src);
1802     gst_object_unref(This->their_sink);
1803     This->my_src = This->their_sink = NULL;
1804
1805     for (i = 0; i < This->cStreams; i++) {
1806         hr = BaseOutputPinImpl_BreakConnect(&ppOldPins[i]->pin);
1807         TRACE("Disconnect: %08x\n", hr);
1808         IPin_Release((IPin*)ppOldPins[i]);
1809     }
1810     This->cStreams = 0;
1811     This->ppPins = NULL;
1812     gst_object_unref(This->gstfilter);
1813     This->gstfilter = NULL;
1814     BaseFilterImpl_IncrementPinVersion((BaseFilter*)This);
1815     CoTaskMemFree(ppOldPins);
1816     return S_OK;
1817 }
1818
1819 static ULONG WINAPI GSTInPin_Release(IPin *iface)
1820 {
1821     GSTInPin *This = (GSTInPin*)iface;
1822     ULONG refCount = InterlockedDecrement(&This->pin.refCount);
1823
1824     TRACE("(%p)->() Release from %d\n", iface, refCount + 1);
1825     if (!refCount) {
1826         FreeMediaType(&This->pin.mtCurrent);
1827         if (This->pAlloc)
1828             IMemAllocator_Release(This->pAlloc);
1829         This->pAlloc = NULL;
1830         This->pin.IPin_iface.lpVtbl = NULL;
1831         return 0;
1832     } else
1833         return refCount;
1834 }
1835
1836 static HRESULT WINAPI GSTInPin_ReceiveConnection(IPin *iface, IPin *pReceivePin, const AM_MEDIA_TYPE *pmt)
1837 {
1838     PIN_DIRECTION pindirReceive;
1839     HRESULT hr = S_OK;
1840     GSTInPin *This = (GSTInPin*)iface;
1841
1842     TRACE("(%p/%p)->(%p, %p)\n", This, iface, pReceivePin, pmt);
1843     dump_AM_MEDIA_TYPE(pmt);
1844
1845     mark_wine_thread();
1846
1847     EnterCriticalSection(This->pin.pCritSec);
1848     if (!This->pin.pConnectedTo) {
1849         ALLOCATOR_PROPERTIES props;
1850
1851         props.cBuffers = 8;
1852         props.cbBuffer = 16384;
1853         props.cbAlign = 1;
1854         props.cbPrefix = 0;
1855
1856         if (SUCCEEDED(hr) && IPin_QueryAccept(iface, pmt) != S_OK)
1857             hr = VFW_E_TYPE_NOT_ACCEPTED;
1858         if (SUCCEEDED(hr)) {
1859             IPin_QueryDirection(pReceivePin, &pindirReceive);
1860             if (pindirReceive != PINDIR_OUTPUT) {
1861                 ERR("Can't connect from non-output pin\n");
1862                 hr = VFW_E_INVALID_DIRECTION;
1863             }
1864         }
1865
1866         This->pReader = NULL;
1867         This->pAlloc = NULL;
1868         if (SUCCEEDED(hr))
1869             hr = IPin_QueryInterface(pReceivePin, &IID_IAsyncReader, (LPVOID *)&This->pReader);
1870         if (SUCCEEDED(hr))
1871             hr = GST_Connect(This, pReceivePin, &props);
1872         if (SUCCEEDED(hr))
1873             hr = IAsyncReader_RequestAllocator(This->pReader, NULL, &props, &This->pAlloc);
1874         if (SUCCEEDED(hr)) {
1875             CopyMediaType(&This->pin.mtCurrent, pmt);
1876             This->pin.pConnectedTo = pReceivePin;
1877             IPin_AddRef(pReceivePin);
1878             hr = IMemAllocator_Commit(This->pAlloc);
1879             SetEvent(((GSTImpl*)This->pin.pinInfo.pFilter)->event);
1880         } else {
1881             GST_RemoveOutputPins((GSTImpl *)This->pin.pinInfo.pFilter);
1882             if (This->pReader)
1883                 IAsyncReader_Release(This->pReader);
1884             This->pReader = NULL;
1885             if (This->pAlloc)
1886                 IMemAllocator_Release(This->pAlloc);
1887             This->pAlloc = NULL;
1888         }
1889         TRACE("Size: %i\n", props.cbBuffer);
1890     } else
1891         hr = VFW_E_ALREADY_CONNECTED;
1892     LeaveCriticalSection(This->pin.pCritSec);
1893     return hr;
1894 }
1895
1896 static HRESULT WINAPI GSTInPin_Disconnect(IPin *iface)
1897 {
1898     HRESULT hr;
1899     GSTInPin *This = (GSTInPin*)iface;
1900     FILTER_STATE state;
1901
1902     TRACE("(%p)\n", This);
1903
1904     mark_wine_thread();
1905
1906     hr = IBaseFilter_GetState(This->pin.pinInfo.pFilter, INFINITE, &state);
1907     EnterCriticalSection(This->pin.pCritSec);
1908     if (This->pin.pConnectedTo) {
1909         GSTImpl *Parser = (GSTImpl *)This->pin.pinInfo.pFilter;
1910
1911         if (SUCCEEDED(hr) && state == State_Stopped) {
1912             IMemAllocator_Decommit(This->pAlloc);
1913             IPin_Disconnect(This->pin.pConnectedTo);
1914             This->pin.pConnectedTo = NULL;
1915             hr = GST_RemoveOutputPins(Parser);
1916         } else
1917             hr = VFW_E_NOT_STOPPED;
1918     } else
1919         hr = S_FALSE;
1920     LeaveCriticalSection(This->pin.pCritSec);
1921     return hr;
1922 }
1923
1924 static HRESULT WINAPI GSTInPin_QueryAccept(IPin *iface, const AM_MEDIA_TYPE *pmt)
1925 {
1926     GSTInPin *This = (GSTInPin*)iface;
1927
1928     TRACE("(%p)->(%p)\n", This, pmt);
1929     dump_AM_MEDIA_TYPE(pmt);
1930
1931     if (IsEqualIID(&pmt->majortype, &MEDIATYPE_Stream))
1932         return S_OK;
1933     return S_FALSE;
1934 }
1935
1936 static HRESULT WINAPI GSTInPin_EndOfStream(IPin *iface)
1937 {
1938     GSTInPin *pin = (GSTInPin*)iface;
1939     GSTImpl *This = (GSTImpl*)pin->pin.pinInfo.pFilter;
1940
1941     FIXME("Propagate message on %p\n", This);
1942     return S_OK;
1943 }
1944
1945 static HRESULT WINAPI GSTInPin_BeginFlush(IPin *iface)
1946 {
1947     GSTInPin *pin = (GSTInPin*)iface;
1948     GSTImpl *This = (GSTImpl*)pin->pin.pinInfo.pFilter;
1949
1950     FIXME("Propagate message on %p\n", This);
1951     return S_OK;
1952 }
1953
1954 static HRESULT WINAPI GSTInPin_EndFlush(IPin *iface)
1955 {
1956     GSTInPin *pin = (GSTInPin*)iface;
1957     GSTImpl *This = (GSTImpl*)pin->pin.pinInfo.pFilter;
1958
1959     FIXME("Propagate message on %p\n", This);
1960     return S_OK;
1961 }
1962
1963 static HRESULT WINAPI GSTInPin_NewSegment(IPin *iface, REFERENCE_TIME tStart, REFERENCE_TIME tStop, double dRate)
1964 {
1965     GSTInPin *pin = (GSTInPin*)iface;
1966     GSTImpl *This = (GSTImpl*)pin->pin.pinInfo.pFilter;
1967
1968     BasePinImpl_NewSegment(iface, tStart, tStop, dRate);
1969     FIXME("Propagate message on %p\n", This);
1970     return S_OK;
1971 }
1972
1973 static HRESULT WINAPI GSTInPin_QueryInterface(IPin * iface, REFIID riid, LPVOID * ppv)
1974 {
1975     GSTInPin *This = (GSTInPin*)iface;
1976
1977     TRACE("(%p/%p)->(%s, %p)\n", This, iface, debugstr_guid(riid), ppv);
1978
1979     *ppv = NULL;
1980
1981     if (IsEqualIID(riid, &IID_IUnknown))
1982         *ppv = iface;
1983     else if (IsEqualIID(riid, &IID_IPin))
1984         *ppv = iface;
1985     else if (IsEqualIID(riid, &IID_IMediaSeeking))
1986     {
1987         return IBaseFilter_QueryInterface(This->pin.pinInfo.pFilter, &IID_IMediaSeeking, ppv);
1988     }
1989
1990     if (*ppv)
1991     {
1992         IUnknown_AddRef((IUnknown *)(*ppv));
1993         return S_OK;
1994     }
1995
1996     FIXME("No interface for %s!\n", debugstr_guid(riid));
1997
1998     return E_NOINTERFACE;
1999 }
2000
2001 static HRESULT WINAPI GSTInPin_EnumMediaTypes(IPin *iface, IEnumMediaTypes **ppEnum)
2002 {
2003     BasePin *This = (BasePin *)iface;
2004
2005     TRACE("(%p/%p)->(%p)\n", This, iface, ppEnum);
2006
2007     return EnumMediaTypes_Construct(This, BasePinImpl_GetMediaType, BasePinImpl_GetMediaTypeVersion, ppEnum);
2008 }
2009
2010 static const IPinVtbl GST_InputPin_Vtbl = {
2011     GSTInPin_QueryInterface,
2012     BasePinImpl_AddRef,
2013     GSTInPin_Release,
2014     BaseInputPinImpl_Connect,
2015     GSTInPin_ReceiveConnection,
2016     GSTInPin_Disconnect,
2017     BasePinImpl_ConnectedTo,
2018     BasePinImpl_ConnectionMediaType,
2019     BasePinImpl_QueryPinInfo,
2020     BasePinImpl_QueryDirection,
2021     BasePinImpl_QueryId,
2022     GSTInPin_QueryAccept,
2023     GSTInPin_EnumMediaTypes,
2024     BasePinImpl_QueryInternalConnections,
2025     GSTInPin_EndOfStream,
2026     GSTInPin_BeginFlush,
2027     GSTInPin_EndFlush,
2028     GSTInPin_NewSegment
2029 };
2030
2031 pthread_mutex_t cb_list_lock = PTHREAD_MUTEX_INITIALIZER;
2032 pthread_cond_t cb_list_cond = PTHREAD_COND_INITIALIZER;
2033 struct list cb_list = LIST_INIT(cb_list);
2034
2035 void CALLBACK perform_cb(TP_CALLBACK_INSTANCE *instance, void *user)
2036 {
2037     struct cb_data *cbdata = user;
2038
2039     TRACE("got cb type: 0x%x\n", cbdata->type);
2040
2041     switch(cbdata->type)
2042     {
2043     case WATCH_BUS:
2044         {
2045             struct watch_bus_data *data = &cbdata->u.watch_bus_data;
2046             cbdata->u.watch_bus_data.ret = watch_bus(data->bus, data->msg, data->user);
2047             break;
2048         }
2049     case EXISTING_NEW_PAD:
2050         {
2051             struct existing_new_pad_data *data = &cbdata->u.existing_new_pad_data;
2052             existing_new_pad(data->bin, data->pad, data->user);
2053             break;
2054         }
2055     case QUERY_FUNCTION:
2056         {
2057             struct query_function_data *data = &cbdata->u.query_function_data;
2058             cbdata->u.query_function_data.ret = query_function(data->pad, data->parent, data->query);
2059             break;
2060         }
2061     case ACTIVATE_MODE:
2062         {
2063             struct activate_mode_data *data = &cbdata->u.activate_mode_data;
2064             cbdata->u.activate_mode_data.ret = activate_mode(data->pad, data->parent, data->mode, data->activate);
2065             break;
2066         }
2067     case NO_MORE_PADS:
2068         {
2069             struct no_more_pads_data *data = &cbdata->u.no_more_pads_data;
2070             no_more_pads(data->decodebin, data->user);
2071             break;
2072         }
2073     case REQUEST_BUFFER_SRC:
2074         {
2075             struct request_buffer_src_data *data = &cbdata->u.request_buffer_src_data;
2076             cbdata->u.request_buffer_src_data.ret = request_buffer_src(data->pad, data->parent,
2077                     data->ofs, data->len, data->buf);
2078             break;
2079         }
2080     case EVENT_SRC:
2081         {
2082             struct event_src_data *data = &cbdata->u.event_src_data;
2083             cbdata->u.event_src_data.ret = event_src(data->pad, data->parent, data->event);
2084             break;
2085         }
2086     case EVENT_SINK:
2087         {
2088             struct event_sink_data *data = &cbdata->u.event_sink_data;
2089             cbdata->u.event_sink_data.ret = event_sink(data->pad, data->parent, data->event);
2090             break;
2091         }
2092     case ACCEPT_CAPS_SINK:
2093         {
2094             struct accept_caps_sink_data *data = &cbdata->u.accept_caps_sink_data;
2095             cbdata->u.accept_caps_sink_data.ret = accept_caps_sink(data->pad, data->caps);
2096             break;
2097         }
2098     case SETCAPS_SINK:
2099         {
2100             struct setcaps_sink_data *data = &cbdata->u.setcaps_sink_data;
2101             cbdata->u.setcaps_sink_data.ret = setcaps_sink(data->pad, data->caps);
2102             break;
2103         }
2104     case GOT_DATA_SINK:
2105         {
2106             struct got_data_sink_data *data = &cbdata->u.got_data_sink_data;
2107             cbdata->u.got_data_sink_data.ret = got_data_sink(data->pad, data->parent, data->buf);
2108             break;
2109         }
2110     case GOT_DATA:
2111         {
2112             struct got_data_data *data = &cbdata->u.got_data_data;
2113             cbdata->u.got_data_data.ret = got_data(data->pad, data->parent, data->buf);
2114             break;
2115         }
2116     case REMOVED_DECODED_PAD:
2117         {
2118             struct removed_decoded_pad_data *data = &cbdata->u.removed_decoded_pad_data;
2119             removed_decoded_pad(data->bin, data->pad, data->user);
2120             break;
2121         }
2122     case AUTOPLUG_BLACKLIST:
2123         {
2124             struct autoplug_blacklist_data *data = &cbdata->u.autoplug_blacklist_data;
2125             cbdata->u.autoplug_blacklist_data.ret = autoplug_blacklist(data->bin,
2126                     data->pad, data->caps, data->fact, data->user);
2127             break;
2128         }
2129     case UNKNOWN_TYPE:
2130         {
2131             struct unknown_type_data *data = &cbdata->u.unknown_type_data;
2132             unknown_type(data->bin, data->pad, data->caps, data->user);
2133             break;
2134         }
2135     case RELEASE_SAMPLE:
2136         {
2137             struct release_sample_data *data = &cbdata->u.release_sample_data;
2138             release_sample(data->data);
2139             break;
2140         }
2141     case TRANSFORM_PAD_ADDED:
2142         {
2143             struct transform_pad_added_data *data = &cbdata->u.transform_pad_added_data;
2144             Gstreamer_transform_pad_added(data->filter, data->pad, data->user);
2145             break;
2146         }
2147     case QUERY_SINK:
2148         {
2149             struct query_sink_data *data = &cbdata->u.query_sink_data;
2150             cbdata->u.query_sink_data.ret = query_sink(data->pad, data->parent,
2151                     data->query);
2152             break;
2153         }
2154     }
2155
2156     pthread_mutex_lock(&cbdata->lock);
2157     cbdata->finished = 1;
2158     pthread_cond_broadcast(&cbdata->cond);
2159     pthread_mutex_unlock(&cbdata->lock);
2160 }
2161
2162 static DWORD WINAPI dispatch_thread(void *user)
2163 {
2164     struct cb_data *cbdata;
2165
2166     pthread_mutex_lock(&cb_list_lock);
2167
2168     while(1){
2169         pthread_cond_wait(&cb_list_cond, &cb_list_lock);
2170
2171         while(!list_empty(&cb_list)){
2172             cbdata = LIST_ENTRY(list_head(&cb_list), struct cb_data, entry);
2173             list_remove(&cbdata->entry);
2174
2175             TrySubmitThreadpoolCallback(&perform_cb, cbdata, NULL);
2176         }
2177     }
2178
2179     pthread_mutex_unlock(&cb_list_lock);
2180
2181     return 0;
2182 }
2183
2184 void start_dispatch_thread(void)
2185 {
2186     pthread_key_create(&wine_gst_key, NULL);
2187     CloseHandle(CreateThread(NULL, 0, &dispatch_thread, NULL, 0, NULL));
2188 }