def _process(pipeline, timeout_ms):
    bus = pipeline.get_bus()
    tags = {}
    mime = None
    have_audio = False
    missing_message = None
    duration = None

    types = (
        Gst.MessageType.ELEMENT |
        Gst.MessageType.APPLICATION |
        Gst.MessageType.ERROR |
        Gst.MessageType.EOS |
        Gst.MessageType.ASYNC_DONE |
        Gst.MessageType.DURATION_CHANGED |
        Gst.MessageType.TAG
    )

    timeout = timeout_ms
    start = int(time.time() * 1000)
    while timeout > 0:
        message = bus.timed_pop_filtered(timeout * Gst.MSECOND, types)

        if message is None:
            break
        elif message.type == Gst.MessageType.ELEMENT:
            if GstPbutils.is_missing_plugin_message(message):
                missing_message = message
        elif message.type == Gst.MessageType.APPLICATION:
            if message.get_structure().get_name() == 'have-type':
                mime = message.get_structure().get_value('caps').get_name()
                if mime and (
                        mime.startswith('text/') or mime == 'application/xml'):
                    return tags, mime, have_audio, duration
            elif message.get_structure().get_name() == 'have-audio':
                have_audio = True
        elif message.type == Gst.MessageType.ERROR:
            error = encoding.locale_decode(message.parse_error()[0])
            if missing_message and not mime:
                caps = missing_message.get_structure().get_value('detail')
                mime = caps.get_structure(0).get_name()
                return tags, mime, have_audio, duration
            raise exceptions.ScannerError(error)
        elif message.type == Gst.MessageType.EOS:
            return tags, mime, have_audio, duration
        elif message.type == Gst.MessageType.ASYNC_DONE:
            success, duration = pipeline.query_duration(Gst.Format.TIME)
            if success:
                duration = duration // Gst.MSECOND
            else:
                duration = None

            if tags and duration is not None:
                return tags, mime, have_audio, duration

            # Workaround for upstream bug which causes tags/duration to arrive
            # after pre-roll. We get around this by starting to play the track
            # and then waiting for a duration change.
            # https://bugzilla.gnome.org/show_bug.cgi?id=763553
            result = pipeline.set_state(Gst.State.PLAYING)
            if result == Gst.StateChangeReturn.FAILURE:
                return tags, mime, have_audio, duration

        elif message.type == Gst.MessageType.DURATION_CHANGED:
            # duration will be read after ASYNC_DONE received; for now
            # just give it a non-None value to flag that we have a duration:
            duration = 0
        elif message.type == Gst.MessageType.TAG:
            taglist = message.parse_tag()
            # Note that this will only keep the last tag.
            tags.update(tags_lib.convert_taglist(taglist))

        timeout = timeout_ms - (int(time.time() * 1000) - start)

        # workaround for https://bugzilla.gnome.org/show_bug.cgi?id=763553:
        # if we got what we want then stop playing (and wait for ASYNC_DONE)
        if tags and duration is not None:
            pipeline.set_state(Gst.State.PAUSED)

    raise exceptions.ScannerError('Timeout after %dms' % timeout_ms)
dataset/ETHPy150Open mopidy/mopidy/mopidy/audio/scan.py/_process/original