| 195 | |
| 196 | class Queue(BaseComponent): |
| 197 | def __init__(self, context, name: str, role: str = "producer"): |
| 198 | endpoint = EndpointManager.get_endpoint(name) |
| 199 | super().__init__(context, endpoint) |
| 200 | |
| 201 | self.name = name |
| 202 | self.role = Role(role) |
| 203 | self.copy = endpoint.copy |
| 204 | self.socket_conf = EndpointManager.config.socket_config |
| 205 | self._msg_id = 0 |
| 206 | |
| 207 | full_ep = f"{endpoint.protocol}://{endpoint.address}" |
| 208 | |
| 209 | self.socket = self.context.socket(zmq.PUSH if self.role == Role.PRODUCER else zmq.PULL) |
| 210 | self.socket_conf.apply(self.socket, self.role == Role.PRODUCER) |
| 211 | |
| 212 | if self.role == Role.PRODUCER: |
| 213 | self.socket.connect(full_ep) |
| 214 | else: |
| 215 | self.socket.bind(full_ep) |
| 216 | |
| 217 | fmq_logger.info(f"Queue {name}({role}) initialized on {full_ep}") |
| 218 | |
| 219 | async def put(self, data: Any, shm_threshold: int = 1024 * 1024): |
| 220 | """ |