(self, fd: Union[int, _Selectable], events: int)
| 173 | self.writers.add(fd) |
| 174 | |
| 175 | def update_handler(self, fd: Union[int, _Selectable], events: int) -> None: |
| 176 | fd, fileobj = self.split_fd(fd) |
| 177 | if events & IOLoop.READ: |
| 178 | if fd not in self.readers: |
| 179 | self.selector_loop.add_reader(fd, self._handle_events, fd, IOLoop.READ) |
| 180 | self.readers.add(fd) |
| 181 | else: |
| 182 | if fd in self.readers: |
| 183 | self.selector_loop.remove_reader(fd) |
| 184 | self.readers.remove(fd) |
| 185 | if events & IOLoop.WRITE: |
| 186 | if fd not in self.writers: |
| 187 | self.selector_loop.add_writer(fd, self._handle_events, fd, IOLoop.WRITE) |
| 188 | self.writers.add(fd) |
| 189 | else: |
| 190 | if fd in self.writers: |
| 191 | self.selector_loop.remove_writer(fd) |
| 192 | self.writers.remove(fd) |
| 193 | |
| 194 | def remove_handler(self, fd: Union[int, _Selectable]) -> None: |
| 195 | fd, fileobj = self.split_fd(fd) |
no test coverage detected