| 165 | self.pidlock.release() |
| 166 | |
| 167 | def setup_queues(self, include, exclude=None): |
| 168 | include = str_to_list(include) |
| 169 | exclude = str_to_list(exclude) |
| 170 | try: |
| 171 | self.app.amqp.queues.select(include) |
| 172 | except KeyError as exc: |
| 173 | raise ImproperlyConfigured( |
| 174 | SELECT_UNKNOWN_QUEUE.strip().format(include, exc)) |
| 175 | try: |
| 176 | self.app.amqp.queues.deselect(exclude) |
| 177 | except KeyError as exc: |
| 178 | raise ImproperlyConfigured( |
| 179 | DESELECT_UNKNOWN_QUEUE.strip().format(exclude, exc)) |
| 180 | if self.app.conf.worker_direct: |
| 181 | self.app.amqp.queues.select_add(worker_direct(self.hostname)) |
| 182 | |
| 183 | def setup_includes(self, includes): |
| 184 | # Update celery_include to have all known task modules, so that we |