(self)
| 50 | super().__init__(*args, **kwargs) |
| 51 | |
| 52 | def run(self): |
| 53 | global THREAD_EVENT |
| 54 | |
| 55 | while not THREAD_EVENT.is_set(): |
| 56 | checked_buckets = len(self.q.checked_buckets) |
| 57 | |
| 58 | if checked_buckets > 1: |
| 59 | cprint("{0} buckets checked ({1:.0f}b/s), {2} buckets found".format( |
| 60 | checked_buckets, |
| 61 | (checked_buckets - self.checked_buckets_since_last_update) / UPDATE_INTERVAL, |
| 62 | FOUND_COUNT), "cyan") |
| 63 | |
| 64 | self.checked_buckets_since_last_update = checked_buckets |
| 65 | THREAD_EVENT.wait(UPDATE_INTERVAL) |
| 66 | |
| 67 | |
| 68 | class CertStreamThread(Thread): |
nothing calls this directly
no outgoing calls
no test coverage detected