|
44 | 44 | UNSUPPORTED_SCHEMA_ROUTE_PREFIX = "babelqueue:unsupported-schema-version:" |
45 | 45 |
|
46 | 46 |
|
| 47 | +_INT32_MIN = -(2**31) |
| 48 | +_INT32_MAX = 2**31 - 1 |
| 49 | + |
| 50 | + |
| 51 | +def _amqp_typed(props: Dict[str, Any]) -> Dict[str, Any]: |
| 52 | + """Pin out-of-int32 ints to AMQP ``long`` (§4.3: ``bq-created-at`` is Unix ms). |
| 53 | +
|
| 54 | + The pure-Python AMQP stack in ``azure-servicebus`` encodes every bare Python int as an |
| 55 | + AMQP int32, so a millisecond epoch overflows at send time. A value outside the int32 range |
| 56 | + is wrapped in the stack's explicit type descriptor; everything else is left as-is. |
| 57 | + """ |
| 58 | + try: |
| 59 | + from azure.servicebus._pyamqp.types import TYPE, VALUE, AMQPTypes |
| 60 | + |
| 61 | + long_type: Any = AMQPTypes.long |
| 62 | + except ImportError: # pragma: no cover - older SDK layouts |
| 63 | + TYPE, VALUE, long_type = "TYPE", "VALUE", "LONG" |
| 64 | + out: Dict[str, Any] = {} |
| 65 | + for key, value in props.items(): |
| 66 | + if ( |
| 67 | + isinstance(value, int) |
| 68 | + and not isinstance(value, bool) |
| 69 | + and not _INT32_MIN <= value <= _INT32_MAX |
| 70 | + ): |
| 71 | + out[key] = {TYPE: long_type, VALUE: value} |
| 72 | + else: |
| 73 | + out[key] = value |
| 74 | + return out |
| 75 | + |
| 76 | + |
47 | 77 | class AsbTransport(Transport): |
48 | 78 | def __init__( |
49 | 79 | self, |
@@ -196,7 +226,10 @@ def _unsupported_schema_version(properties: Any) -> Optional[str]: |
196 | 226 | def publish(self, queue: str, body: str) -> None: |
197 | 227 | from azure.servicebus import ServiceBusMessage |
198 | 228 |
|
199 | | - message = ServiceBusMessage(body, **self._projection(body)) |
| 229 | + kwargs = self._projection(body) |
| 230 | + if kwargs.get("application_properties"): |
| 231 | + kwargs["application_properties"] = _amqp_typed(kwargs["application_properties"]) |
| 232 | + message = ServiceBusMessage(body, **kwargs) |
200 | 233 | self._sender(queue).send_messages(message) |
201 | 234 |
|
202 | 235 | def pop(self, queue: str, timeout: float = 1.0) -> Optional[ReceivedMessage]: |
|
0 commit comments