Media foundation test

mf.cpp

#include <condition_variable>
#include <mutex>
#include <stdexcept>
#define COBJMACROS
#include <initguid.h>
#include <sstream>
#include <mfobjects.h>
#include <mfapi.h>
#include <mfidl.h>
#include <mferror.h>
#include <stdio.h>
#include <memory>
#include <atomic>
DEFINE_GUID(GUID_NULL, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0);
static inline const char *dbgstr_guid( const GUID *id )
{
    static char guid_buf[39];
    if (!id) return "(null)";
    if (!((ULONG_PTR)id >> 16)) {
        snprintf(guid_buf, sizeof(guid_buf), "<guid-0x%04hx>", (WORD)(ULONG_PTR)id );
    } else {
        snprintf(guid_buf, sizeof(guid_buf), "{%08x-%04x-%04x-%02x%02x-%02x%02x%02x%02x%02x%02x}",
                             (unsigned int)id->Data1, id->Data2, id->Data3,
                             id->Data4[0], id->Data4[1], id->Data4[2], id->Data4[3],
                             id->Data4[4], id->Data4[5], id->Data4[6], id->Data4[7] );
    }
    return guid_buf;
}
template <typename... Args>
void ok(bool cond, const char *fmt, Args ...args) {
    char buf[1024];
    if (!cond) {
        snprintf(buf, sizeof(buf), fmt, args...);
        fprintf(stderr, "error: %s\n", buf);
        throw std::runtime_error(buf);
    }
}
#define trace(fmt, ...) fprintf(stderr, fmt, ##__VA_ARGS__)
static IMFMediaSession *session = nullptr;
struct Callback final : public IMFAsyncCallback {
private:
    int ref = 1;
    CRITICAL_SECTION cs;
public:
    Callback() {
        InitializeCriticalSection(&cs);
    }
    ~Callback() {
        DeleteCriticalSection(&cs);
    }
    HRESULT __stdcall QueryInterface(REFIID riid, void **obj) override {
        if (!obj) {
            return E_INVALIDARG;
        }
        if (riid == IID_IMFAsyncCallback) {
            *(IMFAsyncCallback **)obj = this;
            AddRef();
            return S_OK;
        }
        *obj = nullptr;
        return E_NOINTERFACE;
    }
    ULONG __stdcall AddRef() override {
        EnterCriticalSection(&cs);
        int ret = ++ref;
        LeaveCriticalSection(&cs);
        return ret;
    }
    ULONG __stdcall Release() override {
        EnterCriticalSection(&cs);
        int ret = --ref;
        LeaveCriticalSection(&cs);
        return ret;
    }
    HRESULT __stdcall Invoke(IMFAsyncResult *result) override {
        IUnknown *obj = nullptr;
        result->GetObject(&obj);
        fprintf(stderr, "Test %p\n", obj);

        IMFMediaEvent *event = nullptr;
        session->EndGetEvent(result, &event);
        if (obj) {
            obj->Release();
        }
        if (event) {
            MediaEventType type;
            event->GetType(&type);
            fprintf(stderr, "Event %lu\n", type);
            if (type == MESessionTopologyStatus) {
                UINT32 status;
                HRESULT hr = event->GetUINT32(MF_EVENT_TOPOLOGY_STATUS, &status);
                ok(hr == S_OK, "Unexpected hr %#lx.\n", hr);
                trace("status %u\n", status);
            } else if (type == MESessionTopologySet) {
                PROPVARIANT pv;
                event->GetValue(&pv);
                if (pv.vt == VT_EMPTY) {
                    trace("topology set, empty\n");
                } else if (pv.vt == VT_UNKNOWN) {
                    trace("topology set %p\n", pv.punkVal);
                    pv.punkVal->Release();
                }
            }
            event->Release();
        }
        session->BeginGetEvent(this, NULL);
        return S_OK;
    }
    HRESULT __stdcall GetParameters(DWORD *flags, DWORD *queue) override {
        *flags = 0;
        *queue = MFASYNC_CALLBACK_QUEUE_STANDARD;
        return S_OK;
    }
};
static std::mutex m;
static std::condition_variable cv;
static int sample_count = 0;
struct Sink final : public IMFMediaSink, public IMFStreamSink, public IMFClockStateSink {
private:
    int ref = 1;
    IMFPresentationClock *clock;
    IMFMediaEventQueue *queue;
    CRITICAL_SECTION cs;
    std::atomic<bool> shutdown = false;
    std::atomic<bool> clock_running = false;
public:
    IMFMediaTypeHandler *handler;
    Sink() {
        clock = nullptr;
        queue = nullptr;
        if (!SUCCEEDED(MFCreateEventQueue(&queue))) {
            throw std::bad_alloc();
        }

        InitializeCriticalSection(&cs);
        MFCreateSimpleTypeHandler(&handler);
        //QueueEvent(MEStreamSinkRequestSample, GUID_NULL, S_OK, NULL);
    }
    ~Sink() {
        SetPresentationClock(nullptr);
        queue->Release();
        handler->Release();
        DeleteCriticalSection(&cs);
    }
    HRESULT __stdcall QueryInterface(REFIID riid, void **obj) override {
        trace("QueryInterface %p, %p, %s\n", this, obj, dbgstr_guid(&riid));
        if (!obj) {
            return E_INVALIDARG;
        }
        if (riid == IID_IMFMediaSink) {
            *(IMFMediaSink **)obj = this;
            AddRef();
            trace("media sink\n");
            return S_OK;
        } else if (riid == IID_IMFStreamSink) {
            *(IMFStreamSink **)obj = this;
            AddRef();
            trace("stream sink\n");
            return S_OK;
        } else if (riid == IID_IMFMediaEventGenerator) {
            *(IMFMediaEventGenerator **)obj = this;
            this->AddRef();
            trace("event gen\n");
            return S_OK;
        }
        trace("unknown interface\n");
        *obj = nullptr;
        return E_NOINTERFACE;
    }
    ULONG __stdcall AddRef() override {
        EnterCriticalSection(&cs);
        int ret = ++ref;
        LeaveCriticalSection(&cs);
        return ret;
    }
    ULONG __stdcall Release() override {
        EnterCriticalSection(&cs);
        int ret = --ref;
        LeaveCriticalSection(&cs);
        return ret;
    }
    HRESULT __stdcall GetCharacteristics(DWORD *flags) override {
        trace("GetCharacteristics\n");
        *flags = MEDIASINK_FIXED_STREAMS | MEDIASINK_RATELESS;
        return S_OK;
    }
    HRESULT __stdcall AddStreamSink(DWORD id, IMFMediaType *type, IMFStreamSink **sink) override {
        return MF_E_STREAMSINKS_FIXED;
    }
    HRESULT __stdcall RemoveStreamSink(DWORD id) override {
        return MF_E_STREAMSINKS_FIXED;
    }
    HRESULT __stdcall GetStreamSinkCount(DWORD *count) override {
        trace("GetStreamSinkCount\n");
        *count = 1;
        return S_OK;
    }
    HRESULT __stdcall GetStreamSinkByIndex(DWORD index, IMFStreamSink **sink) override {
        trace("GetStreamSinkByIndex %p, %lu\n", this, index);
        if (index != 0) {
            return MF_E_INVALIDINDEX;
        }
        *sink = this;
        AddRef();
        return S_OK;
    }
    HRESULT __stdcall GetStreamSinkById(DWORD id, IMFStreamSink **sink) override {
        if (id != 0) {
            return MF_E_INVALIDSTREAMNUMBER;
        }
        *sink = this;
        AddRef();
        return S_OK;
    }
    HRESULT __stdcall SetPresentationClock(IMFPresentationClock *clock) override {
        trace("SetPresentationClock %p\n", clock);
        if (this->clock) {
            this->clock->RemoveClockStateSink(this);
            this->clock->Release();
        }
        this->clock = clock;
        if (clock) {
            clock->AddRef();
            clock->AddClockStateSink(this);
        }
        return S_OK;
    }
    HRESULT __stdcall GetPresentationClock(IMFPresentationClock **clock) override {
        if (!clock) {
            return E_INVALIDARG;
        }
        if (!this->clock) {
            return MF_E_NO_CLOCK;
        }
        *clock = this->clock;
        (*clock)->AddRef();
        return S_OK;
    }
    HRESULT __stdcall GetEvent(DWORD flags, IMFMediaEvent **event) override {
        trace("GetEvent\n");
        return queue->GetEvent(flags, event);
    }
    HRESULT __stdcall BeginGetEvent(IMFAsyncCallback *callback, IUnknown *state) override {
        if (shutdown.load()) {
            trace("BeginGetEvent after shutdown\n");
            return MF_E_SHUTDOWN;
        }
        trace("BeginGetEvent\n");
        return queue->BeginGetEvent(callback, state);
    }
    HRESULT __stdcall EndGetEvent(IMFAsyncResult *result, IMFMediaEvent **event) override {
        if (shutdown.load()) {
            trace("EndGetEvent after shutdown\n");
            return MF_E_SHUTDOWN;
        }
        HRESULT hr = queue->EndGetEvent(result, event);
        MediaEventType type;
        (*event)->GetType(&type);
        trace("EndGetEvent %lu\n", type);
        return hr;
    }
    HRESULT __stdcall QueueEvent(MediaEventType type, REFGUID ext, HRESULT status, const PROPVARIANT *value) override {
        return queue->QueueEventParamVar(type, ext, status, value);
    }
    HRESULT __stdcall GetMediaSink(IMFMediaSink **sink) override {
        *sink = this;
        AddRef();
        return S_OK;
    }
    HRESULT __stdcall GetIdentifier(DWORD *id) override {
        *id = 0;
        return S_OK;
    }
    HRESULT __stdcall GetMediaTypeHandler(IMFMediaTypeHandler **handler) override {
        trace("GetMediaTypeHandler\n");
        if (!shutdown.load()) {
            this->handler->AddRef();
            *handler = this->handler;
            return S_OK;
        }
        return MF_E_SHUTDOWN;
    }
    HRESULT __stdcall ProcessSample(IMFSample *sample) override {
        trace("ProcessSample\n");
        {
            std::lock_guard<std::mutex> lock(m);
            sample_count++;
            if (sample_count == 100) {
                cv.notify_one();
            }
        }

        if (clock_running.load()) {
            QueueEvent(MEStreamSinkRequestSample, GUID_NULL, S_OK, NULL);
        }
        sample->Release();
        return S_OK;
    }
    HRESULT __stdcall PlaceMarker(MFSTREAMSINK_MARKER_TYPE type, const PROPVARIANT *value, const PROPVARIANT *context) override {
        return S_OK;
    }
    HRESULT __stdcall Flush() override {
        return S_OK;
    }
    HRESULT __stdcall Shutdown() override {
        trace("Shutdown %p\n", this);
        bool f = false;
        if (shutdown.compare_exchange_strong(f, true)) {
            queue->Shutdown();
            return S_OK;
        } else {
            return MF_E_SHUTDOWN;
        }
    }

    /* IMFClockStateSink methods */
    HRESULT __stdcall OnClockStart(MFTIME systime, LONGLONG offset) override {
        trace("OnClockStart\n");
        clock_running.store(true);
        QueueEvent(MEStreamSinkStarted, GUID_NULL, S_OK, NULL);
        QueueEvent(MEStreamSinkRequestSample, GUID_NULL, S_OK, NULL);
        return S_OK;
    }
    HRESULT __stdcall OnClockStop(MFTIME systime) override {
        trace("OnClockStop\n");
        clock_running.store(false);
        QueueEvent(MEStreamSinkStopped, GUID_NULL, S_OK, NULL);
        return S_OK;
    }
    HRESULT __stdcall OnClockPause(MFTIME systime) override {
        trace("OnClockPause\n");
        return S_OK;
    }
    HRESULT __stdcall OnClockRestart(MFTIME systime) override {
        trace("OnClockRestart\n");
        return S_OK;
    }
    HRESULT __stdcall OnClockSetRate(MFTIME systime, float rate) override {
        trace("OnClockSetRate\n");
        return S_OK;
    }

};
static Sink sink[5];
int sink_cnt = 0;
class SinkActivate final : public IMFActivate {
private:
    std::atomic<int> ref = 1;
    IMFMediaSink *sink = NULL;
    IMFAttributes *attr = NULL;
    int id;
public:
    SinkActivate(int id_) : id(id_) {
        if (FAILED(MFCreateAttributes(&attr, 0))) {
            throw std::bad_alloc();
        }
    }
    HRESULT __stdcall QueryInterface(REFIID riid, void **obj) override {
        if (!obj) {
            return E_INVALIDARG;
        }
        if (riid == IID_IMFActivate) {
            *(IMFActivate **)obj = this;
            AddRef();
            return S_OK;
        }
        *obj = nullptr;
        return E_NOINTERFACE;
    }
    ULONG __stdcall AddRef() override {
        return ref.fetch_add(1) + 1;
    }
    ULONG __stdcall Release() override {
        int ret = ref.fetch_sub(1);
        if (ret == 1) {
            delete this;
        }
        return ret - 1;
    }
    HRESULT __stdcall ActivateObject(REFIID riid, void **obj) override {
        if (!obj) {
            return E_INVALIDARG;
        }
        fprintf(stderr, "ActivateObject %p\n", obj);
        if (riid == IID_IMFMediaSink) {
            if (!sink) {
                sink = &::sink[id];
                sink->AddRef();
            }
            sink->AddRef();
            *obj = sink;
            return S_OK;
        }
        *obj = nullptr;
        return E_NOINTERFACE;
    }
    HRESULT __stdcall ShutdownObject() override {
        trace("ShutdownObject\n");
        if (sink) {
            sink->Shutdown();
            sink->Release();
            sink = nullptr;
        }
        return S_OK;
    }
    HRESULT __stdcall DetachObject() override {
        trace("DetachObject\n");
        if (sink) {
            sink->Release();
            sink = nullptr;
        }
        return S_OK;
    }
    /* IMFAttributes methods */
    HRESULT __stdcall Compare(IMFAttributes *theirs, MF_ATTRIBUTES_MATCH_TYPE type, BOOL *result) override {
        return attr->Compare(theirs, type, result);
    }
    HRESULT __stdcall CompareItem(REFGUID key, REFPROPVARIANT value, BOOL *result) override {
        return attr->CompareItem(key, value, result);
    }
    HRESULT __stdcall GetUINT32(REFGUID key, UINT32 *value) override {
        return attr->GetUINT32(key, value);
    }
    HRESULT __stdcall GetUINT64(REFGUID key, UINT64 *value) override {
        return attr->GetUINT64(key, value);
    }
    HRESULT __stdcall GetDouble(REFGUID key, double *value) override {
        return attr->GetDouble(key, value);
    }
    HRESULT __stdcall GetGUID(REFGUID key, GUID *value) override {
        return attr->GetGUID(key, value);
    }
    HRESULT __stdcall GetStringLength(REFGUID key, UINT32 *length) override {
        return attr->GetStringLength(key, length);
    }
    HRESULT __stdcall GetString(REFGUID key, LPWSTR value, UINT32 size, UINT32 *length) override {
        return attr->GetString(key, value, size, length);
    }
    HRESULT __stdcall GetAllocatedString(REFGUID key, LPWSTR *value, UINT32 *length) override {
        return attr->GetAllocatedString(key, value, length);
    }
    HRESULT __stdcall GetBlobSize(REFGUID key, UINT32 *size) override {
        return attr->GetBlobSize(key, size);
    }
    HRESULT __stdcall GetBlob(REFGUID key, UINT8 *buf, UINT32 bufsize, UINT32 *blobsize) override {
        return attr->GetBlob(key, buf, bufsize, blobsize);
    }
    HRESULT __stdcall GetAllocatedBlob(REFGUID key, UINT8 **buf, UINT32 *size) override {
        return attr->GetAllocatedBlob(key, buf, size);
    }
    HRESULT __stdcall GetUnknown(REFGUID key, REFIID riid, void **obj) override {
        return attr->GetUnknown(key, riid, obj);
    }
    HRESULT __stdcall SetItem(REFGUID key, REFPROPVARIANT value) override {
        return attr->SetItem(key, value);
    }
    HRESULT __stdcall DeleteItem(REFGUID key) override {
        return attr->DeleteItem(key);
    }
    HRESULT __stdcall DeleteAllItems() override {
        return attr->DeleteAllItems();
    }
    HRESULT __stdcall SetUINT32(REFGUID key, UINT32 value) override {
        return attr->SetUINT32(key, value);
    }
    HRESULT __stdcall SetUINT64(REFGUID key, UINT64 value) override {
        return attr->SetUINT64(key, value);
    }
    HRESULT __stdcall SetDouble(REFGUID key, double value) override {
        return attr->SetDouble(key, value);
    }
    HRESULT __stdcall SetGUID(REFGUID key, REFGUID value) override {
        return attr->SetGUID(key, value);
    }
    HRESULT __stdcall SetString(REFGUID key, LPCWSTR value) override {
        return attr->SetString(key, value);
    }
    HRESULT __stdcall SetBlob(REFGUID key, const UINT8 *buf, UINT32 size) override {
        return attr->SetBlob(key, buf, size);
    }
    HRESULT __stdcall SetUnknown(REFGUID key, IUnknown *unknown) override {
        return attr->SetUnknown(key, unknown);
    }
    HRESULT __stdcall LockStore() override {
        return attr->LockStore();
    }
    HRESULT __stdcall UnlockStore() override {
        return attr->UnlockStore();
    }
    HRESULT __stdcall GetCount(UINT32 *count) override {
        return attr->GetCount(count);
    }
    HRESULT __stdcall GetItemByIndex(UINT32 index, GUID *key, PROPVARIANT *value) override {
        return attr->GetItemByIndex(index, key, value);
    }
    HRESULT __stdcall CopyAllItems(IMFAttributes *dest) override {
        return attr->CopyAllItems(dest);
    }
    HRESULT __stdcall GetItem(REFGUID key, PROPVARIANT *value) override {
        return attr->GetItem(key, value);
    }
    HRESULT __stdcall GetItemType(REFGUID key, MF_ATTRIBUTE_TYPE *type) override {
        return attr->GetItemType(key, type);
    }

};
HRESULT CreateMediaSource(PCWSTR sURL, IMFMediaSource **ppSource)
{
    MF_OBJECT_TYPE ObjectType = MF_OBJECT_INVALID;

    IMFSourceResolver* pSourceResolver = NULL;
    IUnknown* pSource = NULL;

    // Create the source resolver.
    HRESULT hr = MFCreateSourceResolver(&pSourceResolver);
    if (FAILED(hr))
    {
        goto done;
    }

    // Use the source resolver to create the media source.

    // Note: For simplicity this sample uses the synchronous method to create
    // the media source. However, creating a media source can take a noticeable
    // amount of time, especially for a network source. For a more responsive
    // UI, use the asynchronous BeginCreateObjectFromURL method.

    hr = pSourceResolver->CreateObjectFromURL(
        sURL,                       // URL of the source.
        MF_RESOLUTION_MEDIASOURCE,  // Create a source object.
        NULL,                       // Optional property store.
        &ObjectType,        // Receives the created object type.
        &pSource            // Receives a pointer to the media source.
        );
    if (FAILED(hr))
    {
        goto done;
    }

    // Get the IMFMediaSource interface from the media source.
    hr = pSource->QueryInterface(IID_IMFMediaSource, (void**)ppSource);

done:
    if (pSourceResolver) {
        pSourceResolver->Release();
    }
    if (pSource) {
        pSource->Release();
    }
    return hr;
}

#define SafeRelease(x) if (x) { x->Release(); x = NULL; }
//  Create an activation object for a renderer, based on the stream media type.

void CreateMediaSinkActivate(
    IMFStreamDescriptor *pSourceSD,     // Pointer to the stream descriptor.
    IMFActivate **ppActivate
)
{
    IMFMediaTypeHandler *pHandler = NULL;
    IMFActivate *pActivate = NULL;

    // Get the media type handler for the stream.
    HRESULT hr = pSourceSD->GetMediaTypeHandler(&pHandler);
    ok(hr == S_OK, "Unexpected hr %#lx.\n", hr);

    // Get the major media type.
    GUID guidMajorType;
    hr = pHandler->GetMajorType(&guidMajorType);
    ok(hr == S_OK, "Unexpected hr %#lx.\n", hr);

    IMFMediaType *media_type = NULL;
    hr = pHandler->GetCurrentMediaType(&media_type);
    ok(hr == S_OK, "Unexpected hr %#lx.\n", hr);
    sink[sink_cnt].handler->SetCurrentMediaType(media_type);
    media_type->Release();
    pActivate = new SinkActivate(sink_cnt++);

    // Return IMFActivate pointer to caller.
    *ppActivate = pActivate;
    (*ppActivate)->AddRef();

    SafeRelease(pHandler);
    SafeRelease(pActivate);
}
//  Add a topology branch for one stream.
//
//  For each stream, this function does the following:
//
//    1. Creates a source node associated with the stream.
//    2. Creates an output node for the renderer.
//    3. Connects the two nodes.
//
//  The media session will add any decoders that are needed.

HRESULT AddBranchToPartialTopology(
    IMFTopology *pTopology,         // Topology.
    IMFMediaSource *pSource,        // Media source.
    IMFPresentationDescriptor *pPD, // Presentation descriptor.
    DWORD iStream)                 // Window for video playback.
{
    IMFStreamDescriptor *pSD = NULL;
    IMFActivate         *pSinkActivate = NULL;
    IMFTopologyNode     *pSourceNode = NULL;
    IMFTopologyNode     *pOutputNode = NULL;

    BOOL fSelected = FALSE;

    HRESULT hr = pPD->GetStreamDescriptorByIndex(iStream, &fSelected, &pSD);
    if (FAILED(hr))
    {
        goto done;
    }

    if (fSelected)
    {
        // Create the media sink activation object.
        CreateMediaSinkActivate(pSD, &pSinkActivate);

        hr = MFCreateTopologyNode(MF_TOPOLOGY_SOURCESTREAM_NODE, &pSourceNode);
        if (FAILED(hr)) {
            goto done;
        }

        hr = pSourceNode->SetUnknown(MF_TOPONODE_PRESENTATION_DESCRIPTOR, (IUnknown *)pPD);
        ok(hr == S_OK, "Failed to set node pd, hr %#lx.\n", hr);
        hr = pSourceNode->SetUnknown(MF_TOPONODE_STREAM_DESCRIPTOR, (IUnknown *)pSD);
        ok(hr == S_OK, "Failed to set node sd, hr %#lx.\n", hr);

        hr = pSourceNode->SetUnknown(MF_TOPONODE_SOURCE, (IUnknown *)pSource);
        ok(hr == S_OK, "Unexpected hr %#lx.\n", hr);

        hr = pTopology->AddNode(pSourceNode);
        ok(hr == S_OK, "Failed to add node, hr %#lx.\n", hr);

        trace("adding sink node\n");
        hr = MFCreateTopologyNode(MF_TOPOLOGY_OUTPUT_NODE, &pOutputNode);
        ok(hr == S_OK, "Failed to create topology node, hr %#lx.\n", hr);
        hr = pOutputNode->SetObject(pSinkActivate);
        ok(hr == S_OK, "Failed to set object, hr %#lx.\n", hr);
        hr = pOutputNode->SetUINT32(MF_TOPONODE_RATELESS, 1);
        ok(hr == S_OK, "Failed to set rateless, hr %#lx.\n", hr);
        hr = pTopology->AddNode(pOutputNode);
        ok(hr == S_OK, "Failed to add node, hr %#lx.\n", hr);
        trace("done adding sink node\n");

        // Connect the source node to the output node.
        hr = pSourceNode->ConnectOutput(0, pOutputNode, 0);
        ok(hr == S_OK, "Failed to connect nodes, hr %#lx.\n", hr);
    }
    // else: If not selected, don't add the branch.

done:
    SafeRelease(pSD);
    SafeRelease(pSinkActivate);
    SafeRelease(pSourceNode);
    SafeRelease(pOutputNode);
    return hr;
}

int main() {
    if (!SUCCEEDED(MFStartup(MF_VERSION, MFSTARTUP_FULL))) {
        fprintf(stderr, "MFStartUp\n");
        return 1;
    }
    // IMFMediaEventQueue *q = NULL;
    // if (!SUCCEEDED(MFCreateEventQueue(&q))) {
    //     fprintf(stderr, "MFCreateEventQueue failed\n");
    //     return 1;
    // }
    // Callback cb = Callback();
    // IMFMediaEvent *event;
    // if (!SUCCEEDED(MFCreateMediaEvent(MEError, GUID_NULL, S_OK, NULL, &event))) {
    //     fprintf(stderr, "failed to create event\n");
    //     return 1;
    // }
    // q->BeginGetEvent(&cb, NULL);
    // q->QueueEvent(event);
    // for(;getchar(););
    MFCreateMediaSession(NULL, &session);

    const wchar_t url[] = L"http://192.168.1.156:8000/a.mp4";
    IMFMediaSource *source = nullptr;
    HRESULT hr = CreateMediaSource(url, &source);
    Callback cb{};
    session->BeginGetEvent(&cb, NULL);
    if (FAILED(hr)) {
        std::ostringstream err{};
        err << "failed to create media source " << std::hex << hr;
        throw std::runtime_error(err.str());
    }

    IMFTopology *pTopology = NULL;
    IMFPresentationDescriptor *pPD = NULL;
    DWORD cSourceStreams = 0;

    if (FAILED(source->CreatePresentationDescriptor(&pPD))) {
        throw std::runtime_error("failed to create presentation descriptor");
    }

    // Create a new topology.
    hr = MFCreateTopology(&pTopology);
    if (FAILED(hr))
    {
        throw std::runtime_error("failed to create topology");
    }

    // Get the number of streams in the media source.
    hr = pPD->GetStreamDescriptorCount(&cSourceStreams);
    if (FAILED(hr))
    {
        throw std::runtime_error("failed to get stream descriptor count");
    }

    // For each stream, create the topology nodes and add them to the topology.
    for (DWORD i = 0; i < cSourceStreams; i++)
    {
        fprintf(stderr, "Adding branch %d\n", i);
        hr = AddBranchToPartialTopology(pTopology, source, pPD, i);
        if (FAILED(hr))
        {
            throw std::runtime_error("failed to add branch to partial topology");
        }
    }

    hr = session->SetTopology(0, pTopology);
    ok(hr == S_OK, "Failed to set topology, hr %#lx.\n", hr);

    PROPVARIANT pv;
    pv.vt = VT_EMPTY;
    hr = session->Start(&GUID_NULL, &pv);
    ok(hr == S_OK, "Failed to start session, hr %#lx.\n", hr);
    while (true) {
        std::unique_lock<std::mutex> lock(m);
        if (sample_count >= 100) {
            trace("sample_count %d, stop\n", sample_count);
            break;
        }
        cv.wait(lock);
    }

    session->Shutdown();
    session->Release();
    Sleep(1000);
    return 0;
}