| 247 | self._msg_id += 1 |
| 248 | |
| 249 | async def get(self, timeout: int = None) -> Optional[Message]: |
| 250 | # Receive data from queue |
| 251 | if self.role != Role.CONSUMER: |
| 252 | raise PermissionError("Only consumers can get messages.") |
| 253 | |
| 254 | try: |
| 255 | if timeout: |
| 256 | raw = await asyncio.wait_for(self.socket.recv(), timeout / 1000) |
| 257 | else: |
| 258 | raw = await self.socket.recv(copy=self.copy) |
| 259 | except asyncio.TimeoutError: |
| 260 | fmq_logger.error(f"Timeout receiving message on {self.name}") |
| 261 | return None |
| 262 | |
| 263 | msg = Message.deserialize(raw) |
| 264 | if msg.descriptor: |
| 265 | msg.payload = msg.descriptor.read_and_unlink() |
| 266 | |
| 267 | self._msg_id += 1 |
| 268 | return msg |
| 269 | |
| 270 | |
| 271 | # ========================== |