Args: ---------- num_samples : optional positive integer Maximal number of messages to take from the queue, if set to None all available messages will be taken. The call blocks until there are any messages available. It may return less tha
(self, num_samples=1, predicate=None)
| 152 | return msgs_len |
| 153 | |
| 154 | def get(self, num_samples=1, predicate=None) -> Optional[List[MSG_CLASS]]: |
| 155 | """ |
| 156 | Args: |
| 157 | ---------- |
| 158 | num_samples : optional positive integer |
| 159 | Maximal number of messages to take from the queue, if set to None all available messages |
| 160 | will be taken. The call blocks until there are any messages available. |
| 161 | It may return less than `num_samples`, but an empty list is returned only if `predicate` |
| 162 | was specified and it evaluated to False after waiting on empty queue. |
| 163 | The call returns None iff the queue was closed. |
| 164 | predicate : a parameterless callable |
| 165 | Used for double-checking if the item should really be taken after waiting on empty |
| 166 | queue. |
| 167 | """ |
| 168 | if self.is_closed: |
| 169 | return |
| 170 | with self.cv_not_empty: # equivalent to `with self.lock` |
| 171 | waited = self._wait_for_samples() |
| 172 | if self.meta.is_closed: |
| 173 | self.is_closed = True |
| 174 | self.cv_not_empty.notify() |
| 175 | return |
| 176 | if waited and predicate is not None and not predicate(): |
| 177 | recv = [] |
| 178 | else: |
| 179 | recv = self._recv_samples(num_samples) |
| 180 | if self.meta.size > 0: |
| 181 | self.cv_not_empty.notify() |
| 182 | return recv |
| 183 | |
| 184 | |
| 185 | class Dispatcher: |
no test coverage detected