| 131 | self.cv_not_empty.notify() |
| 132 | |
| 133 | def put(self, msgs: List[MSG_CLASS]) -> Optional[int]: |
| 134 | assert len(msgs), "Cannot write an empty list of messages" |
| 135 | if self.is_closed: |
| 136 | return |
| 137 | with self.lock: |
| 138 | self._read_meta() |
| 139 | if self.meta.size + len(msgs) > self.meta.capacity: |
| 140 | raise RuntimeError("The queue is full") |
| 141 | if self.meta.is_closed: |
| 142 | self.is_closed = True |
| 143 | return |
| 144 | msgs_len = len(msgs) |
| 145 | next_slot = (self.meta.begining + self.meta.size) % self.meta.capacity |
| 146 | for msg in msgs: |
| 147 | self._write_msg(next_slot, msg) |
| 148 | next_slot = (next_slot + 1) % self.meta.capacity |
| 149 | self.meta.size += msgs_len |
| 150 | self._write_meta() |
| 151 | self.cv_not_empty.notify() |
| 152 | return msgs_len |
| 153 | |
| 154 | def get(self, num_samples=1, predicate=None) -> Optional[List[MSG_CLASS]]: |
| 155 | """ |