Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/zmq/sugar/socket.py: 47%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""0MQ Socket pure Python methods."""
3# Copyright (C) PyZMQ Developers
4# Distributed under the terms of the Modified BSD License.
6from __future__ import annotations
8import errno
9import pickle
10import random
11import sys
12from collections.abc import Sequence
13from typing import (
14 TYPE_CHECKING,
15 Any,
16 Callable,
17 Generic,
18 Literal,
19 TypeVar,
20 cast,
21 overload,
22)
23from warnings import warn
25import zmq
26from zmq._typing import TypeAlias
27from zmq.backend import Socket as SocketBase
28from zmq.error import ZMQBindError, ZMQError
29from zmq.utils import jsonapi
30from zmq.utils.interop import cast_int_addr
32from ..constants import SocketOption, SocketType, _OptType
33from .attrsettr import AttributeSetter
34from .poll import Poller
36if TYPE_CHECKING:
37 from typing_extensions import Buffer, Self
39try:
40 DEFAULT_PROTOCOL = pickle.DEFAULT_PROTOCOL
41except AttributeError:
42 DEFAULT_PROTOCOL = pickle.HIGHEST_PROTOCOL
44_T = TypeVar("_T")
45_RT = TypeVar("_RT")
46_SocketT_co = TypeVar("_SocketT_co", bound="Socket", covariant=True)
48# must match the `jsonapi.loads` return type
49_JSON: TypeAlias = "dict[str, Any] | list[Any] | str | float"
52class _SocketContext(Generic[_SocketT_co]):
53 """Context Manager for socket bind/unbind"""
55 socket: _SocketT_co
56 kind: str
57 addr: str
59 def __repr__(self):
60 return f"<SocketContext({self.kind}={self.addr!r})>"
62 def __init__(self, socket: _SocketT_co, kind: str, addr: str):
63 assert kind in {"bind", "connect"}
64 self.socket = socket
65 self.kind = kind
66 self.addr = addr
68 def __enter__(self) -> _SocketT_co:
69 return self.socket
71 def __exit__(self, *args):
72 if self.socket.closed:
73 return
74 if self.kind == "bind":
75 self.socket.unbind(self.addr)
76 elif self.kind == "connect":
77 self.socket.disconnect(self.addr)
80_SocketReturnT_co = TypeVar("_SocketReturnT_co", covariant=True)
83class Socket(SocketBase, AttributeSetter, Generic[_SocketReturnT_co]):
84 """The ZMQ socket object
86 To create a Socket, first create a Context::
88 ctx = zmq.Context.instance()
90 then call ``ctx.socket(socket_type)``::
92 s = ctx.socket(zmq.ROUTER)
94 .. versionadded:: 25
96 Sockets can now be shadowed by passing another Socket.
97 This helps in creating an async copy of a sync socket or vice versa::
99 s = zmq.Socket(async_socket)
101 Which previously had to be::
103 s = zmq.Socket.shadow(async_socket.underlying)
104 """
106 _shadow = False
107 _shadow_obj: zmq.Socket | int | None = None
108 _monitor_socket = None
109 _type_name = 'UNKNOWN'
111 context: zmq.Context
113 @overload
114 def __init__(
115 self: Socket[bytes],
116 ctx_or_socket: zmq.Context,
117 socket_type: int,
118 *,
119 copy_threshold: int | None = None,
120 ) -> None: ...
121 @overload
122 def __init__(
123 self: Socket[bytes],
124 *,
125 shadow: Socket | int,
126 copy_threshold: int | None = None,
127 ) -> None: ...
128 @overload
129 def __init__(
130 self: Socket[bytes],
131 ctx_or_socket: Socket,
132 ) -> None: ...
133 def __init__(
134 self: Socket[bytes],
135 ctx_or_socket: zmq.Context | Socket | None = None,
136 socket_type: int = 0,
137 *,
138 shadow: Socket | int = 0,
139 copy_threshold: int | None = None,
140 ) -> None:
141 shadow_context: zmq.Context | None = None
142 if isinstance(ctx_or_socket, zmq.Socket):
143 # positional Socket(other_socket)
144 shadow = ctx_or_socket
145 ctx_or_socket = None
147 shadow_address: int = 0
149 if shadow:
150 self._shadow = True
151 # hold a reference to the shadow object
152 self._shadow_obj = shadow
153 if not isinstance(shadow, int):
154 if isinstance(shadow, zmq.Socket):
155 shadow_context = shadow.context
156 try:
157 shadow = shadow.underlying
158 except AttributeError:
159 pass
160 shadow_address = cast_int_addr(shadow)
161 else:
162 self._shadow = False
164 super().__init__(
165 ctx_or_socket,
166 socket_type,
167 shadow=shadow_address,
168 copy_threshold=copy_threshold,
169 )
170 if self._shadow_obj and shadow_context:
171 # keep self.context reference if shadowing a Socket object
172 self.context = shadow_context
174 try:
175 socket_type = cast(int, self.get(zmq.TYPE))
176 except Exception:
177 pass
178 else:
179 try:
180 self.__dict__["type"] = stype = SocketType(socket_type)
181 except ValueError:
182 self._type_name = str(socket_type)
183 else:
184 self._type_name = stype.name
186 def __del__(self) -> None:
187 if not self._shadow and not self.closed:
188 if warn is not None:
189 # warn can be None during process teardown
190 warn(
191 f"Unclosed socket {self}",
192 ResourceWarning,
193 stacklevel=2,
194 source=self,
195 )
196 self.close()
198 _repr_cls = "zmq.Socket"
200 def __repr__(self) -> str:
201 cls = self.__class__
202 # look up _repr_cls on exact class, not inherited
203 _repr_cls = cls.__dict__.get("_repr_cls", None)
204 if _repr_cls is None:
205 _repr_cls = f"{cls.__module__}.{cls.__name__}"
207 closed = ' closed' if self._closed else ''
209 return f"<{_repr_cls}(zmq.{self._type_name}) at {hex(id(self))}{closed}>"
211 # socket as context manager:
212 def __enter__(self) -> Self:
213 """Sockets are context managers
215 .. versionadded:: 14.4
216 """
217 return self
219 def __exit__(self, *args: Any, **kwargs: Any) -> None:
220 self.close()
222 # -------------------------------------------------------------------------
223 # Socket creation
224 # -------------------------------------------------------------------------
226 def __copy__(self, memo: Any | None = None) -> Self:
227 """Copying a Socket creates a shadow copy"""
228 return self.__class__.shadow(self.underlying)
230 __deepcopy__ = __copy__
232 @classmethod
233 def shadow(cls, address: int | zmq.Socket) -> Self:
234 """Shadow an existing libzmq socket
236 address is a zmq.Socket or an integer (or FFI pointer)
237 representing the address of the libzmq socket.
239 .. versionadded:: 14.1
241 .. versionadded:: 25
242 Support for shadowing `zmq.Socket` objects,
243 instead of just integer addresses.
244 """
245 return cls(shadow=address)
247 def close(self, linger: int | None = None) -> None:
248 """
249 Close the socket.
251 If linger is specified, LINGER sockopt will be set prior to closing.
253 Note: closing a zmq Socket may not close the underlying sockets
254 if there are undelivered messages.
255 Only after all messages are delivered or discarded by reaching the socket's LINGER timeout
256 (default: forever)
257 will the underlying sockets be closed.
259 This can be called to close the socket by hand. If this is not
260 called, the socket will automatically be closed when it is
261 garbage collected,
262 in which case you may see a ResourceWarning about the unclosed socket.
263 """
264 if self.context:
265 self.context._rm_socket(self)
266 super().close(linger=linger)
268 # -------------------------------------------------------------------------
269 # Connect/Bind context managers
270 # -------------------------------------------------------------------------
272 def _connect_cm(self, addr: str) -> _SocketContext[Self]:
273 """Context manager to disconnect on exit
275 .. versionadded:: 20.0
276 """
277 return _SocketContext(self, 'connect', addr)
279 def _bind_cm(self, addr: str) -> _SocketContext[Self]:
280 """Context manager to unbind on exit
282 .. versionadded:: 20.0
283 """
284 try:
285 # retrieve last_endpoint
286 # to support binding on random ports via
287 # `socket.bind('tcp://127.0.0.1:0')`
288 addr = cast(bytes, self.get(zmq.LAST_ENDPOINT)).decode("utf8")
289 except (AttributeError, ZMQError, UnicodeDecodeError):
290 pass
291 return _SocketContext(self, 'bind', addr)
293 def bind(self, addr: str) -> _SocketContext[Self]: # type:ignore[override]
294 """s.bind(addr)
296 Bind the socket to an address.
298 This causes the socket to listen on a network port. Sockets on the
299 other side of this connection will use ``Socket.connect(addr)`` to
300 connect to this socket.
302 Returns a context manager which will call unbind on exit.
304 .. versionadded:: 20.0
305 Can be used as a context manager.
307 .. versionadded:: 26.0
308 binding to port 0 can be used as a context manager
309 for binding to a random port.
310 The URL can be retrieved as `socket.last_endpoint`.
312 Parameters
313 ----------
314 addr : str
315 The address string. This has the form 'protocol://interface:port',
316 for example 'tcp://127.0.0.1:5555'. Protocols supported include
317 tcp, udp, pgm, epgm, inproc and ipc. If the address is unicode, it is
318 encoded to utf-8 first.
320 """
321 try:
322 super().bind(addr)
323 except ZMQError as e:
324 e.strerror += f" (addr={addr!r})"
325 raise
326 return self._bind_cm(addr)
328 def connect(self, addr: str) -> _SocketContext[Self]: # type:ignore[override]
329 """s.connect(addr)
331 Connect to a remote 0MQ socket.
333 Returns a context manager which will call disconnect on exit.
335 .. versionadded:: 20.0
336 Can be used as a context manager.
338 Parameters
339 ----------
340 addr : str
341 The address string. This has the form 'protocol://interface:port',
342 for example 'tcp://127.0.0.1:5555'. Protocols supported are
343 tcp, udp, pgm, inproc and ipc. If the address is unicode, it is
344 encoded to utf-8 first.
346 """
347 try:
348 super().connect(addr)
349 except ZMQError as e:
350 e.strerror += f" (addr={addr!r})"
351 raise
352 return self._connect_cm(addr)
354 # -------------------------------------------------------------------------
355 # Deprecated aliases
356 # -------------------------------------------------------------------------
358 @property
359 def socket_type(self) -> int:
360 warn("Socket.socket_type is deprecated, use Socket.type", DeprecationWarning)
361 return cast(int, self.type)
363 # -------------------------------------------------------------------------
364 # Hooks for sockopt completion
365 # -------------------------------------------------------------------------
367 def __dir__(self) -> list[str]:
368 keys = dir(self.__class__)
369 keys.extend(SocketOption.__members__)
370 return keys
372 # -------------------------------------------------------------------------
373 # Getting/Setting options
374 # -------------------------------------------------------------------------
375 setsockopt = SocketBase.set
376 getsockopt = SocketBase.get
378 def __setattr__(self, key: str, value: Any) -> None:
379 """Override to allow setting zmq.[UN]SUBSCRIBE even though we have a subscribe method"""
380 if key in self.__dict__:
381 object.__setattr__(self, key, value)
382 return
383 _key = key.lower()
384 if _key in ('subscribe', 'unsubscribe'):
385 if isinstance(value, str):
386 value = value.encode('utf8')
387 if _key == 'subscribe':
388 self.set(zmq.SUBSCRIBE, value)
389 else:
390 self.set(zmq.UNSUBSCRIBE, value)
391 return
392 super().__setattr__(key, value)
394 def fileno(self) -> int:
395 """Return edge-triggered file descriptor for this socket.
397 This is a read-only edge-triggered file descriptor for both read and write events on this socket.
398 It is important that all available events be consumed when an event is detected,
399 otherwise the read event will not trigger again.
401 .. versionadded:: 17.0
402 """
403 return self.FD
405 def subscribe(self, topic: str | bytes) -> None:
406 """Subscribe to a topic
408 Only for SUB sockets.
410 .. versionadded:: 15.3
411 """
412 if isinstance(topic, str):
413 topic = topic.encode('utf8')
414 self.set(zmq.SUBSCRIBE, topic)
416 def unsubscribe(self, topic: str | bytes) -> None:
417 """Unsubscribe from a topic
419 Only for SUB sockets.
421 .. versionadded:: 15.3
422 """
423 if isinstance(topic, str):
424 topic = topic.encode('utf8')
425 self.set(zmq.UNSUBSCRIBE, topic)
427 def set_string(self, option: int, optval: str, encoding: str = 'utf-8') -> None:
428 """Set socket options with a unicode object.
430 This is simply a wrapper for setsockopt to protect from encoding ambiguity.
432 See the 0MQ documentation for details on specific options.
434 Parameters
435 ----------
436 option : int
437 The name of the option to set. Can be any of: SUBSCRIBE,
438 UNSUBSCRIBE, IDENTITY
439 optval : str
440 The value of the option to set.
441 encoding : str
442 The encoding to be used, default is utf8
443 """
444 if not isinstance(optval, str):
445 raise TypeError(f"strings only, not {type(optval)}: {optval!r}")
446 return self.set(option, optval.encode(encoding))
448 setsockopt_unicode = setsockopt_string = set_string
450 def get_string(self, option: int, encoding: str = 'utf-8') -> str:
451 """Get the value of a socket option.
453 See the 0MQ documentation for details on specific options.
455 Parameters
456 ----------
457 option : int
458 The option to retrieve.
460 Returns
461 -------
462 optval : str
463 The value of the option as a unicode string.
464 """
465 if SocketOption(option)._opt_type != _OptType.bytes:
466 raise TypeError(f"option {option} will not return a string to be decoded")
467 return cast(bytes, self.get(option)).decode(encoding)
469 getsockopt_unicode = getsockopt_string = get_string
471 def bind_to_random_port(
472 self,
473 addr: str,
474 min_port: int = 49152,
475 max_port: int = 65536,
476 max_tries: int = 100,
477 ) -> int:
478 """Bind this socket to a random port in a range.
480 If the port range is unspecified, the system will choose the port.
482 Parameters
483 ----------
484 addr : str
485 The address string without the port to pass to ``Socket.bind()``.
486 min_port : int, optional
487 The minimum port in the range of ports to try (inclusive).
488 max_port : int, optional
489 The maximum port in the range of ports to try (exclusive).
490 max_tries : int, optional
491 The maximum number of bind attempts to make.
493 Returns
494 -------
495 port : int
496 The port the socket was bound to.
498 Raises
499 ------
500 ZMQBindError
501 if `max_tries` reached before successful bind
502 """
503 if min_port == 49152 and max_port == 65536:
504 # if LAST_ENDPOINT is supported, and min_port / max_port weren't specified,
505 # we can bind to port 0 and let the OS do the work
506 self.bind(f"{addr}:*")
507 url = cast(bytes, self.last_endpoint).decode('ascii', 'replace')
508 _, port_s = url.rsplit(':', 1)
509 return int(port_s)
511 for i in range(max_tries):
512 try:
513 port = random.randrange(min_port, max_port)
514 self.bind(f'{addr}:{port}')
515 except ZMQError as exception:
516 en = exception.errno
517 if en == zmq.EADDRINUSE:
518 continue
519 elif sys.platform == 'win32' and en == errno.EACCES:
520 continue
521 else:
522 raise
523 else:
524 return port
525 raise ZMQBindError("Could not bind socket to random port.")
527 def get_hwm(self) -> int:
528 """Get the High Water Mark.
530 On libzmq ≥ 3, this gets SNDHWM if available, otherwise RCVHWM
531 """
532 # return sndhwm, fallback on rcvhwm
533 try:
534 return cast(int, self.get(zmq.SNDHWM))
535 except zmq.ZMQError:
536 pass
538 return cast(int, self.get(zmq.RCVHWM))
540 def set_hwm(self, value: int) -> None:
541 """Set the High Water Mark.
543 On libzmq ≥ 3, this sets both SNDHWM and RCVHWM
546 .. warning::
548 New values only take effect for subsequent socket
549 bind/connects.
550 """
551 raised = None
552 try:
553 self.sndhwm = value
554 except Exception as e:
555 raised = e
556 try:
557 self.rcvhwm = value
558 except Exception as e:
559 raised = e
561 if raised:
562 raise raised
564 if TYPE_CHECKING:
566 @property
567 def hwm(self) -> int: ...
568 @hwm.setter
569 def hwm(self, hwm: int, /) -> None: ...
571 else:
572 hwm: property = property(
573 get_hwm,
574 set_hwm,
575 None,
576 """Property for High Water Mark.
578 Setting hwm sets both SNDHWM and RCVHWM as appropriate.
579 It gets SNDHWM if available, otherwise RCVHWM.
580 """,
581 )
583 # -------------------------------------------------------------------------
584 # Sending and receiving messages
585 # -------------------------------------------------------------------------
587 @overload # type:ignore[override] # data: Buffer, copy=True (default)
588 def send(
589 self,
590 data: Buffer,
591 flags: int = 0,
592 copy: Literal[True] = True,
593 track: bool = False,
594 routing_id: int | None = None,
595 group: str | None = None,
596 ) -> None: ...
597 @overload # data: Buffer, copy=False (keyword)
598 def send(
599 self,
600 data: Buffer,
601 flags: int = 0,
602 *,
603 copy: Literal[False],
604 track: bool = False,
605 routing_id: int | None = None,
606 group: str | None = None,
607 ) -> zmq.MessageTracker: ...
608 @overload # data: Buffer, copy=False (positional)
609 def send(
610 self,
611 data: Buffer,
612 flags: int,
613 copy: Literal[False],
614 track: bool = False,
615 routing_id: int | None = None,
616 group: str | None = None,
617 ) -> zmq.MessageTracker: ...
618 @overload # data: Buffer, copy=True|False (mypy bug workaround)
619 def send(
620 self,
621 data: Buffer,
622 flags: int,
623 copy: bool = True,
624 track: bool = False,
625 routing_id: int | None = None,
626 group: str | None = None,
627 ) -> zmq.MessageTracker | None: ...
628 @overload # data: Frame
629 def send(
630 self,
631 data: zmq.Frame,
632 flags: int = 0,
633 copy: bool = True,
634 track: bool = False,
635 routing_id: int | None = None,
636 group: str | None = None,
637 ) -> zmq.MessageTracker: ...
638 def send(
639 self,
640 data: Any,
641 flags: int = 0,
642 copy: bool = True,
643 track: bool = False,
644 routing_id: int | None = None,
645 group: str | None = None,
646 ) -> zmq.MessageTracker | None:
647 """Send a single zmq message frame on this socket.
649 This queues the message to be sent by the IO thread at a later time.
651 With flags=NOBLOCK, this raises :class:`ZMQError` if the queue is full;
652 otherwise, this waits until space is available.
653 See :class:`Poller` for more general non-blocking I/O.
655 Parameters
656 ----------
657 data : bytes, Frame, memoryview
658 The content of the message. This can be any object that provides
659 the Python buffer API (i.e. `memoryview(data)` can be called).
660 flags : int
661 0, NOBLOCK, SNDMORE, or NOBLOCK|SNDMORE.
662 copy : bool
663 Should the message be sent in a copying or non-copying manner.
664 track : bool
665 Should the message be tracked for notification that ZMQ has
666 finished with it? (ignored if copy=True)
667 routing_id : int
668 For use with SERVER sockets
669 group : str
670 For use with RADIO sockets
672 Returns
673 -------
674 MessageTracker : if `data` is a :class:`Frame` or `copy=False`
675 a MessageTracker object, whose `done` property will
676 be False until the send is completed.
677 None : otherwise
678 None if message was sent, raises an exception otherwise.
680 Raises
681 ------
682 TypeError
683 If a unicode object is passed
684 ValueError
685 If `track=True`, but an untracked Frame is passed.
686 ZMQError
687 If the send does not succeed for any reason (including
688 if NOBLOCK is set and the outgoing queue is full).
691 .. versionchanged:: 17.0
693 DRAFT support for routing_id and group arguments.
694 """
695 if routing_id is not None:
696 if not isinstance(data, zmq.Frame):
697 data = zmq.Frame(
698 data,
699 track=track,
700 copy=copy or None,
701 copy_threshold=self.copy_threshold,
702 )
703 data.routing_id = routing_id
704 if group is not None:
705 if not isinstance(data, zmq.Frame):
706 data = zmq.Frame(
707 data,
708 track=track,
709 copy=copy or None,
710 copy_threshold=self.copy_threshold,
711 )
712 data.group = group
713 return super().send(data, flags=flags, copy=copy, track=track)
715 def send_multipart(
716 self,
717 msg_parts: Sequence[zmq.Frame | Buffer],
718 flags: int = 0,
719 copy: bool = True,
720 track: bool = False,
721 **kwargs: Any,
722 ) -> zmq.MessageTracker | None:
723 """Send a sequence of buffers as a multipart message.
725 The zmq.SNDMORE flag is added to all msg parts before the last.
727 Parameters
728 ----------
729 msg_parts : iterable
730 A sequence of objects to send as a multipart message. Each element
731 can be any sendable object (Frame, bytes, buffer-providers)
732 flags : int, optional
733 Any valid flags for :func:`Socket.send`.
734 SNDMORE is added automatically for frames before the last.
735 copy : bool, optional
736 Should the frame(s) be sent in a copying or non-copying manner.
737 If copy=False, frames smaller than self.copy_threshold bytes
738 will be copied anyway.
739 track : bool, optional
740 Should the frame(s) be tracked for notification that ZMQ has
741 finished with it (ignored if copy=True).
743 Returns
744 -------
745 MessageTracker : if any part is a :class:`Frame` or `copy=False`
746 a MessageTracker object, whose `done` property will
747 be False until the send is completed.
748 None : otherwise
749 None if message was sent, raises an exception otherwise.
750 """
751 # typecheck parts before sending:
752 for i, msg in enumerate(msg_parts):
753 if isinstance(msg, (zmq.Frame, bytes, memoryview)):
754 continue
755 try:
756 memoryview(msg)
757 except Exception:
758 rmsg = repr(msg)
759 if len(rmsg) > 32:
760 rmsg = rmsg[:32] + '...'
761 raise TypeError(
762 f"Frame {i} ({rmsg}) does not support the buffer interface."
763 )
764 for msg in msg_parts[:-1]:
765 self.send(msg, zmq.SNDMORE | flags, copy=copy, track=track)
766 # Send the last part without the extra SNDMORE flag.
767 return self.send(msg_parts[-1], flags, copy=copy, track=track)
769 @overload # copy=True (default)
770 def recv_multipart(
771 self, flags: int = 0, copy: Literal[True] = True, track: bool = False
772 ) -> list[bytes]: ...
773 @overload # copy=False (keyword)
774 def recv_multipart(
775 self, flags: int = 0, *, copy: Literal[False], track: bool = False
776 ) -> list[zmq.Frame]: ...
777 @overload # copy=False (positional)
778 def recv_multipart(
779 self, flags: int, copy: Literal[False], track: bool = False
780 ) -> list[zmq.Frame]: ...
781 @overload # (mypy bug workaround)
782 def recv_multipart(
783 self, flags: int, copy: bool, track: bool = False
784 ) -> list[bytes] | list[zmq.Frame]: ...
785 def recv_multipart(
786 self, flags: int = 0, copy: bool = True, track: bool = False
787 ) -> list[Any]:
788 """Receive a multipart message as a list of bytes or Frame objects
790 Parameters
791 ----------
792 flags : int, optional
793 Any valid flags for :func:`Socket.recv`.
794 copy : bool, optional
795 Should the message frame(s) be received in a copying or non-copying manner?
796 If False a Frame object is returned for each part, if True a copy of
797 the bytes is made for each frame.
798 track : bool, optional
799 Should the message frame(s) be tracked for notification that ZMQ has
800 finished with it? (ignored if copy=True)
802 Returns
803 -------
804 msg_parts : list
805 A list of frames in the multipart message; either Frames or bytes,
806 depending on `copy`.
808 Raises
809 ------
810 ZMQError
811 for any of the reasons :func:`~Socket.recv` might fail
812 """
813 parts = [self.recv(flags, copy=copy, track=track)]
814 # have first part already, only loop while more to receive
815 while self.getsockopt(zmq.RCVMORE):
816 part = self.recv(flags, copy=copy, track=track)
817 parts.append(part)
818 return parts
820 def _deserialize(
821 self,
822 recvd: _T,
823 load: Callable[[_T], _RT],
824 ) -> _RT:
825 """Deserialize a received message
827 Override in subclass (e.g. Futures) if recvd is not the raw bytes.
829 The default implementation expects bytes and returns the deserialized message immediately.
831 Parameters
832 ----------
834 load: callable
835 Callable that deserializes bytes
836 recvd:
837 The object returned by self.recv
839 """
840 return load(recvd)
842 def send_serialized(
843 self,
844 msg: _T,
845 serialize: Callable[[_T], Sequence[zmq.Frame | Buffer]],
846 flags: int = 0,
847 copy: bool = True,
848 **kwargs: Any,
849 ) -> zmq.MessageTracker | None:
850 """Send a message with a custom serialization function.
852 .. versionadded:: 17
854 Parameters
855 ----------
856 msg : The message to be sent. Can be any object serializable by `serialize`.
857 serialize : callable
858 The serialization function to use.
859 serialize(msg) should return an iterable of sendable message frames
860 (e.g. bytes objects), which will be passed to send_multipart.
861 flags : int, optional
862 Any valid flags for :func:`Socket.send`.
863 copy : bool, optional
864 Whether to copy the frames.
866 """
867 frames = serialize(msg)
868 return self.send_multipart(frames, flags=flags, copy=copy, **kwargs)
870 # these overloads should not be needed, but mypy seems to have trouble inferring
871 # the type of deserialize's argument without them
872 @overload
873 def recv_serialized(
874 self,
875 deserialize: Callable[[list[bytes]], _RT],
876 flags: int = 0,
877 copy: bool = True,
878 ) -> _RT: ...
879 @overload
880 def recv_serialized(
881 self,
882 deserialize: Callable[[list[zmq.Frame]], _RT],
883 flags: int = 0,
884 copy: bool = True,
885 ) -> _RT: ...
886 def recv_serialized(
887 self,
888 deserialize: Callable[[list[Any]], _RT],
889 flags: int = 0,
890 copy: bool = True,
891 ) -> _RT:
892 """Receive a message with a custom deserialization function.
894 .. versionadded:: 17
896 Parameters
897 ----------
898 deserialize : callable
899 The deserialization function to use.
900 deserialize will be called with one argument: the list of frames
901 returned by recv_multipart() and can return any object.
902 flags : int, optional
903 Any valid flags for :func:`Socket.recv`.
904 copy : bool, optional
905 Whether to recv bytes or Frame objects.
907 Returns
908 -------
909 obj : object
910 The object returned by the deserialization function.
912 Raises
913 ------
914 ZMQError
915 for any of the reasons :func:`~Socket.recv` might fail
916 """
917 frames = self.recv_multipart(flags=flags, copy=copy)
918 return self._deserialize(frames, deserialize)
920 def send_string(
921 self,
922 u: str,
923 flags: int = 0,
924 copy: bool = True,
925 encoding: str = 'utf-8',
926 **kwargs: Any,
927 ) -> zmq.MessageTracker | None:
928 """Send a Python unicode string as a message with an encoding.
930 0MQ communicates with raw bytes, so you must encode/decode
931 text (str) around 0MQ.
933 Parameters
934 ----------
935 u : str
936 The unicode string to send.
937 flags : int, optional
938 Any valid flags for :func:`Socket.send`.
939 encoding : str
940 The encoding to be used
941 """
942 if not isinstance(u, str):
943 raise TypeError("str objects only")
944 return self.send(u.encode(encoding), flags=flags, copy=copy, **kwargs)
946 send_unicode = send_string
948 def recv_string(self, flags: int = 0, encoding: str = 'utf-8') -> str:
949 """Receive a unicode string, as sent by send_string.
951 Parameters
952 ----------
953 flags : int
954 Any valid flags for :func:`Socket.recv`.
955 encoding : str
956 The encoding to be used
958 Returns
959 -------
960 s : str
961 The Python unicode string that arrives as encoded bytes.
963 Raises
964 ------
965 ZMQError
966 for any of the reasons :func:`Socket.recv` might fail
967 """
968 msg = self.recv(flags=flags)
969 return self._deserialize(msg, lambda buf: buf.decode(encoding))
971 recv_unicode = recv_string
973 def send_pyobj(
974 self,
975 obj: object,
976 flags: int = 0,
977 protocol: int = DEFAULT_PROTOCOL,
978 **kwargs: Any,
979 ) -> zmq.MessageTracker | None:
980 """
981 Send a Python object as a message using pickle to serialize.
983 .. warning::
985 Never deserialize an untrusted message with pickle,
986 which can involve arbitrary code execution.
987 Make sure to authenticate the sources of messages
988 before unpickling them, e.g. with transport-level security
989 (e.g. CURVE, ZAP, or IPC permissions)
990 or signed messages.
992 Parameters
993 ----------
994 obj : Python object
995 The Python object to send.
996 flags : int
997 Any valid flags for :func:`Socket.send`.
998 protocol : int
999 The pickle protocol number to use. The default is pickle.DEFAULT_PROTOCOL
1000 where defined, and pickle.HIGHEST_PROTOCOL elsewhere.
1001 """
1002 msg = pickle.dumps(obj, protocol)
1003 return self.send(msg, flags=flags, **kwargs)
1005 def recv_pyobj(self, flags: int = 0) -> Any:
1006 """
1007 Receive a Python object as a message using UNSAFE pickle to serialize.
1009 .. warning::
1011 Never deserialize an untrusted message with pickle,
1012 which can involve arbitrary code execution.
1013 Make sure to authenticate the sources of messages
1014 before unpickling them, e.g. with transport-level security
1015 (such as CURVE or IPC permissions)
1016 or authenticating messages themselves before deserializing.
1018 Parameters
1019 ----------
1020 flags : int
1021 Any valid flags for :func:`Socket.recv`.
1023 Returns
1024 -------
1025 obj : Python object
1026 The Python object that arrives as a message.
1028 Raises
1029 ------
1030 ZMQError
1031 for any of the reasons :func:`~Socket.recv` might fail
1032 """
1033 msg = self.recv(flags)
1034 return self._deserialize(msg, pickle.loads)
1036 def send_json(self, obj: object, flags: int = 0, **kwargs: Any) -> None:
1037 """Send a Python object as a message using json to serialize.
1039 Keyword arguments are passed on to json.dumps
1041 Parameters
1042 ----------
1043 obj : Python object
1044 The Python object to send
1045 flags : int
1046 Any valid flags for :func:`Socket.send`
1047 """
1048 send_kwargs = {}
1049 for key in ('routing_id', 'group'):
1050 if key in kwargs:
1051 send_kwargs[key] = kwargs.pop(key)
1052 msg = jsonapi.dumps(obj, **kwargs)
1053 return self.send(msg, flags=flags, **send_kwargs)
1055 def recv_json(self, flags: int = 0, **kwargs: Any) -> _JSON:
1056 """Receive a Python object as a message using json to serialize.
1058 Keyword arguments are passed on to json.loads
1060 Parameters
1061 ----------
1062 flags : int
1063 Any valid flags for :func:`Socket.recv`.
1065 Returns
1066 -------
1067 obj : Python object
1068 The Python object that arrives as a message.
1070 Raises
1071 ------
1072 ZMQError
1073 for any of the reasons :func:`~Socket.recv` might fail
1074 """
1075 msg = self.recv(flags)
1076 return self._deserialize(msg, lambda buf: jsonapi.loads(buf, **kwargs))
1078 _poller_class = Poller
1080 def poll(self, timeout: int | None = None, flags: int = zmq.POLLIN) -> int:
1081 """Poll the socket for events.
1083 See :class:`Poller` to wait for multiple sockets at once.
1085 Parameters
1086 ----------
1087 timeout : int
1088 The timeout (in milliseconds) to wait for an event. If unspecified
1089 (or specified None), will wait forever for an event.
1090 flags : int
1091 default: POLLIN.
1092 POLLIN, POLLOUT, or POLLIN|POLLOUT. The event flags to poll for.
1094 Returns
1095 -------
1096 event_mask : int
1097 The poll event mask (POLLIN, POLLOUT),
1098 0 if the timeout was reached without an event.
1099 """
1101 if self.closed:
1102 raise ZMQError(zmq.ENOTSUP)
1104 p = self._poller_class()
1105 p.register(self, flags)
1106 evts = dict(p.poll(timeout))
1107 # return 0 if no events, otherwise return event bitfield
1108 return evts.get(self, 0)
1110 def get_monitor_socket(
1111 self, events: int | None = None, addr: str | None = None
1112 ) -> Self:
1113 """Return a connected PAIR socket ready to receive the event notifications.
1115 .. versionadded:: libzmq-4.0
1116 .. versionadded:: 14.0
1118 Parameters
1119 ----------
1120 events : int
1121 default: `zmq.EVENT_ALL`
1122 The bitmask defining which events are wanted.
1123 addr : str
1124 The optional endpoint for the monitoring sockets.
1126 Returns
1127 -------
1128 socket : zmq.Socket
1129 The PAIR socket, connected and ready to receive messages.
1130 """
1131 # safe-guard, method only available on libzmq >= 4
1132 if zmq.zmq_version_info() < (4,):
1133 raise NotImplementedError(
1134 f"get_monitor_socket requires libzmq >= 4, have {zmq.zmq_version()}"
1135 )
1137 # if already monitoring, return existing socket
1138 if self._monitor_socket:
1139 if self._monitor_socket.closed:
1140 self._monitor_socket = None
1141 else:
1142 return self._monitor_socket
1144 if addr is None:
1145 # create endpoint name from internal fd
1146 addr = f"inproc://monitor.s-{self.FD}"
1147 if events is None:
1148 # use all events
1149 events = zmq.EVENT_ALL
1150 # attach monitoring socket
1151 self.monitor(addr, events)
1152 # create new PAIR socket and connect it
1153 self._monitor_socket = socket = self.context.socket(zmq.PAIR)
1154 socket.connect(addr)
1155 return socket
1157 def disable_monitor(self) -> None:
1158 """Shutdown the PAIR socket (created using get_monitor_socket)
1159 that is serving socket events.
1161 .. versionadded:: 14.4
1162 """
1163 self._monitor_socket = None
1164 self.monitor(None, 0)
1167SyncSocket: TypeAlias = Socket[bytes]
1169__all__ = ['Socket', 'SyncSocket']