Skip to main content

Asynchronous API for ZMQ using AnyIO

Project description

Build Status

zmq-anyio

Asynchronous API for ZMQ using AnyIO.

Usage

zmq_anyio.Socket is a subclass of zmq.Socket. Here is how it must be used:

  • Create a zmq_anyio.Socket from a zmq.Socket or from a zmq.Context:
    • Create a blocking ZMQ socket using a zmq.Context, and pass it to an async zmq_anyio.Socket:
      ctx = zmq.Context()
      sock = ctx.socket(zmq.PAIR)
      asock = zmq_anyio.Socket(sock)
      
    • Or create an async zmq_anyio.Socket using a zmq.Context:
      ctx = zmq.Context()
      asock = zmq_anyio.Socket(ctx)
      
  • Use the zmq_anyio.Socket with an async context manager.
  • Use arecv() for the async API, recv() for the blocking API, etc.
import anyio
import zmq
import zmq_anyio

ctx = zmq.Context()
sock1 = ctx.socket(zmq.PAIR)
port = sock1.bind("tcp://127.0.0.1:1234")
sock2 = ctx.socket(zmq.PAIR)
sock2.connect("tcp://127.0.0.1:1234")

# wrap the `zmq.Socket` with `zmq_anyio.Socket`:
sock1 = zmq_anyio.Socket(sock1)
sock2 = zmq_anyio.Socket(sock2)

async def main():
    async with sock1, sock2:  # use an async context manager
        await sock1.asend(b"Hello").wait()  # use `asend` instead of `send`, and await the `.wait()` method
        sock1.asend(b", World!")  # or don't await it, it's sent in the background
        assert await sock2.arecv().wait() == b"Hello"  # use `arecv` instead of `recv`, and await the `.wait()` method
        future = sock2.arecv()  # or get the future and await it later
        assert await future.wait() == b", World!"

anyio.run(main)

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

zmq_anyio-0.3.9.tar.gz (13.2 kB view details)

Uploaded Source

Built Distribution

zmq_anyio-0.3.9-py3-none-any.whl (10.5 kB view details)

Uploaded Python 3

File details

Details for the file zmq_anyio-0.3.9.tar.gz.

File metadata

  • Download URL: zmq_anyio-0.3.9.tar.gz
  • Upload date:
  • Size: 13.2 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? Yes
  • Uploaded via: twine/6.1.0 CPython/3.12.9

File hashes

Hashes for zmq_anyio-0.3.9.tar.gz
Algorithm Hash digest
SHA256 7dc0cdbf039834f16c0dc6412b560e689ca406e4d8b93cb94e0fa8763a87e947
MD5 dc6c7e2248f49207d387a3b217c060e7
BLAKE2b-256 4688dcebf3052e2dbe9b81f1d13ad6c1b41eb0e742c7f75a73ce00a78375cb76

See more details on using hashes here.

Provenance

The following attestation bundles were made for zmq_anyio-0.3.9.tar.gz:

Publisher: publish.yml on QuantStack/zmq-anyio

Attestations: Values shown here reflect the state when the release was signed and may no longer be current.

File details

Details for the file zmq_anyio-0.3.9-py3-none-any.whl.

File metadata

  • Download URL: zmq_anyio-0.3.9-py3-none-any.whl
  • Upload date:
  • Size: 10.5 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? Yes
  • Uploaded via: twine/6.1.0 CPython/3.12.9

File hashes

Hashes for zmq_anyio-0.3.9-py3-none-any.whl
Algorithm Hash digest
SHA256 a922b13409ae49cf51730958f0a37b38d9ce9ebde05f046fec9f8639b58ac709
MD5 43551d0bb6c457fdd5f098323f59032c
BLAKE2b-256 7e0b137721c37bbe04d0a379d349342382edc0c48a0feefd52e69c63b2c7bfe9

See more details on using hashes here.

Provenance

The following attestation bundles were made for zmq_anyio-0.3.9-py3-none-any.whl:

Publisher: publish.yml on QuantStack/zmq-anyio

Attestations: Values shown here reflect the state when the release was signed and may no longer be current.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page