is-wire-sea

Middleware AMQP em Python para a arquitetura IS. A distribuição no PyPI se chama is-wire-sea; o pacote Python estável continua sendo is_wire.

Na versão 3.0, remover bindings de filas nomeadas exige unsubscribe(topic, shared_queue=True). Consulte MIGRATION.md antes de atualizar aplicações da série 2.x. O formato wire permanece compatível.

Instalação

python -m pip install is-wire-sea

Há suporte para Python 3.10 a 3.14. Instale um exportador OTLP com:

python -m pip install 'is-wire-sea[tracing]'

Broker

Implantações de produção devem fixar o RabbitMQ pela versão de patch e pelo digest da imagem. A base de desenvolvimento é o RabbitMQ 4.3.1:

docker run --rm -p 5672:5672 \
  rabbitmq:4.3.1@sha256:6a46d2aef889d2a8cc28ac91b4a1ca0116a4a151de10cada219ee7685dc01c5b

RabbitMQ 3.7.6 é apenas um alvo de migração. Mova as implantações existentes da versão 3.7.6 para um cluster 4.3 novo usando uma migração blue-green; não reutilize diretamente o diretório de dados antigo.

Pub/sub

A API existente de entrega no máximo uma vez é preservada:

from is_wire.core import Channel, Message, Subscription

with Channel("amqp://guest:guest@localhost:5672") as channel:
    subscription = Subscription(channel)
    subscription.subscribe("Camera.Config")

    channel.publish(Message(content=b"hello"), topic="Camera.Config")
    received = channel.consume(timeout=1.0)

Assinaturas anônimas são exclusivas e temporárias. Assinaturas nomeadas são duráveis, compartilhadas entre réplicas e expiram após cinco minutos sem uso.

Fluxos de imagens em tempo real

Use StreamSubscription quando quadros antigos puderem ser descartados. Réplicas com o mesmo grupo compartilham uma fila, então cada quadro é processado por uma réplica em vez de ser copiado para todas elas.

from is_msgs.image_pb2 import Image
from is_wire.core import Channel, Message, StreamSubscription


def process(message):
    image = message.unpack(Image)
    # image.data já contém bytes JPEG, WebP ou PNG.


with Channel("amqp://guest:guest@localhost:5672") as channel:
    stream = StreamSubscription(channel, group="detector")
    stream.subscribe("Camera.*.Frame")
    stream.run(process)

Publique um quadro com semântica de fluxo:

image = Image(data=encoded_jpeg_bytes)
channel.publish_stream(Message(content=image), topic="Camera.0.Frame")

O preset de fluxo usa confirmações manuais, prefetch 1, fila com tamanho um, descarte do item mais antigo quando há overflow, TTL de dois segundos e mensagens não persistentes. Uma exceção no callback rejeita o quadro sem recolocá-lo na fila. Use outro grupo somente quando outra etapa de processamento realmente precisar de sua própria cópia.

Image.data deve conter bytes binários codificados da imagem. Não envie RGB/BGR bruto, a menos que o orçamento de rede permita explicitamente; não codifique o corpo em base64 e não aplique compressão gzip/zstd genérica a um payload JPEG, WebP ou PNG já comprimido.

Use um Channel separado para o tráfego de imagens e para o tráfego de RPC/controle, evitando que quadros grandes atrasem mensagens pequenas de controle.

Lifecycle e callbacks

Subscription.close() cancela o consumer e remove o registro no Channel. Também pode ser usada como context manager. Ela mantém filas nomeadas e seus bindings compartilhados; entregas manuais ainda no buffer local são recolocadas na fila. Em uma subscription padrão, mensagens já retornadas por consume() continuam sob responsabilidade do chamador. Fechar uma subscription de stream fecha seu canal AMQP dedicado e pode reentregar frames em flight.

unsubscribe(topic) continua disponível para filas anônimas. Em filas nomeadas, o binding é compartilhado por todas as réplicas: a remoção exige unsubscribe(topic, shared_queue=True) e coordenação entre os processos, inclusive durante reconexões. As assinaturas da mesma conexão são atualizadas juntas; outras conexões não recebem essa atualização de configuração. Para encerrar apenas o consumidor local, use close().

StreamSubscription.stop() sinaliza o loop; não cancela o consumer. O callback executa em uma thread daemon e não pode ser morto com segurança pelo Python. Depois do período de graça, run() rejeita o frame e retorna, mas o callback pode continuar executando. Uma nova chamada a run() é recusada enquanto esse callback anterior estiver vivo. Nunca suponha que rejeitar o frame cancela os efeitos externos do callback. Um grupo admite um frame pendente mais um em flight por consumer; TTL não interrompe trabalho que já começou.

Channel tem um único dono de I/O; múltiplos consumidores não devem executar loops de leitura simultâneos na mesma conexão. Durante callbacks longos, pode ser útil ativar stream.run(process, poll_during_callback=True) para drenar a conexão e atender heartbeats. Essa opção exige que o callback deixe ack/reject para o loop e use um Channel separado, criado na thread do callback, para publicar resultados. O default preserva o agendamento existente. stop() pode ser chamado de outra thread; close() deve ser chamado pelo dono do I/O.

Protobuf

from google.protobuf.struct_pb2 import Struct
from is_wire.core import ContentType, Message

value = Struct()
value["camera"] = "front"

binary = Message(content=value)
assert binary.content_type == ContentType.PROTOBUF
assert binary.unpack(Struct) == value

json_message = Message(content_type=ContentType.JSON)
json_message.pack(value)

Há suporte para Protobuf 5 a 7. Os corpos binários existentes e as convenções de propriedades AMQP continuam compatíveis com is-wire 1.2.1.

Envelopes AMQP incompatíveis com o formato IS, como IDs de correlação não hexadecimais ou cabeçalhos rpc-status inválidos, são ignorados sem interromper o consumo. Entregas com ack manual são rejeitadas sem requeue; uma dead-letter exchange configurada no broker pode recebê-las. O timeout original continua valendo e o aviso de descarte não inclui o corpo nem os cabeçalhos. Isso não altera o tratamento de um payload RPC inválido dentro de um envelope válido.

RPC

from google.protobuf.struct_pb2 import Struct
from is_wire.core import Channel
from is_wire.rpc import ServiceProvider


def echo(request, context):
    return request


channel = Channel("amqp://guest:guest@localhost:5672")
provider = ServiceProvider(channel)
provider.delegate("Echo", echo, Struct, Struct)
provider.run()

As filas de serviços RPC usam confirmações manuais e prefetch 16. A requisição é confirmada depois que basic_publish() retorna. Publicações não usam publisher confirms nem mandatory: isso não comprova aceite ou roteamento da resposta pelo broker. Falhas entre processamento e ack podem reentregar a requisição, e uma resposta pode se perder mesmo após o ack. Os handlers devem ser idempotentes; a semântica atual não é exactly-once e não garante entrega persistente da resposta após reinício do broker.

OpenTelemetry

Tracer, TracingInterceptor, Message.inject_tracing e Message.extract_tracing usam OpenTelemetry e propagação B3 com múltiplos cabeçalhos. IDs de trace legados de 64 bits e IDs de 128 bits são aceitos. Spans aninhados compartilham o trace ativo e restauram o contexto anterior ao terminar, inclusive sem instalar o SDK opcional.

from is_wire.core import Message, Tracer

tracer = Tracer()
with tracer.span("publish") as span:
    message = Message(content=b"payload")
    message.inject_tracing(span)

Zipkin

Instale o exportador nativo OpenTelemetry Zipkin JSON v2:

python -m pip install 'is-wire-sea[zipkin]'

Crie um provider compartilhado por processo. Ele agrupa as exportações em segundo plano, aplica amostragem baseada no contexto pai e identifica cada instância de câmera separadamente no Zipkin:

from is_wire.core import Message, ZipkinTracing
from is_wire.rpc import ServiceProvider, TracingInterceptor

tracing = ZipkinTracing(
    service_name="camera-gateway",
    endpoint="http://zipkin:9411/api/v2/spans",
    service_instance_id="camera-5",
    sample_ratio=0.1,
    resource_attributes={"camera.driver": "hikvision"},
)
frame_tracer = tracing.tracer()

provider = ServiceProvider(rpc_channel)
provider.add_interceptor(TracingInterceptor(tracing=tracing))

with frame_tracer.span("camera.frame") as span:
    span.set_attribute("camera.id", "5")
    message = Message(content=image)
    message.inject_tracing(span)
    stream_channel.publish_stream(message, topic="CameraGateway.5.Frame")

# Libere os spans enfileirados durante o encerramento normal do processo.
tracing.shutdown()

Não associe corpos de imagem, credenciais, URLs de câmeras ou identificadores sem limite aos spans. Use uma proporção de amostragem menor que um para vídeo contínuo e defina 1.0 apenas em sessões curtas de diagnóstico. Consumidores continuam o trace do produtor com message.extract_tracing().

Commtrace opcional

CommTraceClient acrescenta relatórios de comunicação ao tracing existente. Ele envia um JSON por UDP no formato usado pelo is-person-detector. Zipkin/OTLP continuam recebendo os spans de processamento; commtrace recebe os horários e os IDs B3 para correlacionar a comunicação entre serviços. O cliente usa a biblioteca padrão, sem extras ou dependências adicionais.

Para registrar automaticamente mensagens recebidas, configure explicitamente commtrace= no Channel. Isso vale para Subscription, RPC e StreamSubscription:

from is_wire.core import Channel, CommTraceClient, StreamSubscription

with CommTraceClient("127.0.0.1", 9999, span_name="frame") as commtrace:
    with Channel("amqp://guest:guest@localhost:5672", commtrace=commtrace) as channel:
        stream = StreamSubscription(channel, group="detector")
        stream.subscribe("CameraGateway.*.Frame")
        message = stream.consume(timeout=1.0)
        # O relatório é enviado antes de consume() retornar.
        # Processe a imagem e continue o tracing normalmente.
        message.ack()

O endereço e a porta são do seu coletor UDP; 9999 é apenas um exemplo. O produtor precisa incluir B3, por exemplo com message.inject_tracing(span). Sem commtrace=, não há envio nem abertura de socket de commtrace. O cliente pode ser compartilhado entre canais e deve ser fechado pelo chamador; fechar um Channel não fecha o cliente.

Para envio manual, depois de um consumo em canal sem commtrace=, use:

from is_wire.core import CommTraceClient, now

received_at = now()  # Capture imediatamente após receber a mensagem.
with CommTraceClient("127.0.0.1", 9999) as commtrace:
    sent = commtrace.send(message, received_at=received_at, span_name="frame")

O datagrama contém exatamente timestamp_send, timestamp_rcvd, x-b3-flags, x-b3-parentspanid, x-b3-sampled, x-b3-spanid, x-b3-traceid e spanname. Os timestamps são strings com microssegundos Unix; timestamp_send deriva de message.created_at, e timestamp_rcvd é capturado na entrega à aplicação. A diferença inclui filas e espera antes do consumo; ela não isola o tempo de trânsito na rede. Sincronize os relógios dos serviços para comparar horários de máquinas diferentes.

Contextos B3 de 64/128 bits e a decisão sampled=false são preservados. Mensagens sem contexto B3 válido não geram relatórios. O relatório não inclui o corpo nem os demais metadados da mensagem, e não altera sua serialização AMQP.

O socket é reutilizado e o envio é não bloqueante por padrão (timeout=0.0). Falhas UDP/DNS retornam False e não interrompem o consumo AMQP; não há retries. True significa que o socket local aceitou o datagrama, sem confirmação do coletor. O primeiro envio resolve o hostname, e DNS não tem deadline rígido; use um endereço IP quando esse custo importar. Um timeout positivo limita a espera do socket, mas pode acrescentar esse tempo ao consumo automático. UDP pode perder ou reordenar relatórios. Ativar commtrace acrescenta trabalho ao processo; não acelera o AMQP nem substitui o tracing de processamento.

Métricas

As métricas do Prometheus são registradas no registry padrão. MetricsInterceptor.start_server() pode expô-las por HTTP. O perfil 2.0 exporta:

• is_wire_messages_published_total e is_wire_published_bytes_total;
• is_wire_messages_received_total e is_wire_received_bytes_total;
• is_wire_stream_processing_seconds, is_wire_stream_frame_age_seconds e is_wire_stream_callback_errors_total;
• is_wire_reconnections_total;
• is_wire_rpc_duration_seconds e is_wire_rpc_requests_total, particionados por status.

O conteúdo das imagens e os IDs de correlação nunca são usados como labels de métricas. Topics e grupos, porém, são labels: respostas em filas anônimas geram séries com IDs de fila que permanecem no registry. Mantenha grupos estáveis e planeje agrupamento de topics antes de usar essas métricas em tráfego com alta rotatividade de clientes. Os contadores de publicação medem chamadas que retornaram, não confirmações recebidas do broker.

TLS e ciclo de vida da conexão

amqps:// habilita TLS com verificação de certificado e hostname. Opções SSL adicionais do py-amqp podem ser passadas em ssl_options. Consumidores se reconectam com backoff exponencial e jitter; publicações nunca são repetidas automaticamente, pois a entrega pode já ter ocorrido. O deadline de consume(timeout=...) inclui retries e espera de reconexão, e o backoff continua entre chamadas com timeout. A conexão inicial ainda propaga falhas. Os timeouts do py-amqp são por operação: DNS e múltiplos frames do handshake podem ultrapassar o orçamento, portanto não são deadlines rígidos de tempo real. close(timeout=5.0) limita a espera de I/O do encerramento e sempre coleta o transporte local. Erros permanentes de autenticação/certificado são propagados.

channel = Channel(
    "amqps://user:password@rabbitmq.example:5671/vhost",
    heartbeat=30,
    ssl_options={"ca_certs": "/etc/ssl/certs/cluster-ca.pem"},
)

Desenvolvimento

python -m pip install -e '.[dev,tracing]'
pytest  # requer RabbitMQ; use pytest -m "not integration" sem broker
ruff check src tests
python -m build
twine check --strict dist/*

Regenere o schema wire interno de forma reproduzível com o compilador Protobuf 5.29 fixado:

python -m pip install -e '.[codegen]'
python scripts/generate_wire.py

Execute o perfil de publicador a 30 FPS / consumidor a 10 FPS contra um broker de teste com:

python scripts/validate_stream_profile.py --payload-size 1048576 --frames 300

Execute o mesmo comando em implantações Kubernetes separadas que compartilhem um group para comparar as métricas de entrada/saída do RabbitMQ conforme réplicas são adicionadas.

Consulte MIGRATION.md, COMPATIBILITY.md e CHANGELOG.md antes de atualizar uma implantação existente.

Agradecimentos

A equipe do LabVISIO para a base do repositório is-wire-py. Para a modernização, a revisão de compatibilidade, os testes e o empacotamento no PyPI da série 2.0 foram realizados com assistência do OpenAI Codex.