"""Consumer helpers."""

from struct import unpack

from confluent_kafka.schema_registry import _MAGIC_BYTE
from confluent_kafka.schema_registry.avro import _ContextStringIO

from kafka_utils.exceptions import DeserializationException


def read_schema_id_from_message(serialized_msg):
    """Get the schema ID from the serialized message."""
    if len(serialized_msg) <= 5:
        raise DeserializationException('Unrecognized message format')
    with _ContextStringIO(serialized_msg) as payload:
        magic, schema_id = unpack('>bI', payload.read(5))
        if magic != _MAGIC_BYTE:
            raise DeserializationException('Unrecognized message format')
        return schema_id