def METHOD_NAME(self, stream: Iterable[Doc], batch_size: int, **kwargs) -> Union[Generator[Doc, None, None], Iterator[Doc]]:  # type: ignore
    error_handler = self.get_error_handler()
    if kwargs.get("parallel", False):
        PipeRunner._execute, PipeRunner._delayed = self._lazy_init_pool()
        for docs in minibatch(stream, size=self.workers):
            docs = [PipeRunner.serialize_entities(doc) for doc in docs]
            try:
                tasks = (PipeRunner._delayed(self.__call__, doc, Underscore.get_state()) for doc in docs)
                for output_doc in PipeRunner._execute(tasks):
                    yield PipeRunner.deserialize_entities(output_doc)
            except Exception as e:
                error_handler(self.name, self, docs, e)  # type: ignore
                yield from [None] * len(docs)
    else:
        for doc in stream:
            try:
                yield self(doc)
            except Exception as e:
                error_handler(self.name, self, [doc], e)  # type: ignore
                yield None