def __call__(self, config):
tmp_dir_suffix = self.tmp_dir_suffix<EOL>if tmp_dir_suffix is None:<EOL><INDENT>with self.lock:<EOL><INDENT>self.tmp_dir_suffix = str(uuid.uuid4())<EOL>try:<EOL><INDENT>(reader, incremental_transforms, batch_transforms,<EOL>pbi_transforms, writers,<EOL>tmp_dir_path) = self._init_all_stages(config)<EOL><DEDENT>finally:<EOL><INDENT>self.tmp_dir_suffix = None<EOL><DEDENT><DEDENT><DEDENT>else:<EOL><INDENT>(reader, incremental_transforms, batch_transforms,<EOL>pbi_transforms, writers,<EOL>tmp_dir_path) = self._init_all_stages(config)<EOL><DEDENT>return Pipeline(<EOL>rate_log_interval=config['<STR_LIT>'],<EOL>input_item_limit=config.get('<STR_LIT>'),<EOL>cleanup_tmp_files=config['<STR_LIT>'],<EOL>tmp_dir_path=tmp_dir_path,<EOL>assert_single_source=config['<STR_LIT>'],<EOL>output_chunk_max_count=config.get('<STR_LIT>'),<EOL>output_max_clean_visible_bytes=config.get(<EOL>'<STR_LIT>'),<EOL>reader=reader,<EOL>incremental_transforms=incremental_transforms,<EOL>batch_transforms=batch_transforms,<EOL>post_batch_incremental_transforms=pbi_transforms,<EOL>writers=writers,<EOL>)<EOL>
Create a :class:`Pipeline`.

        Pass in the configuration under the ``streamcorpus_pipeline``
        block, not the top-level configuration that contains it.

        If :attr:`tmp_dir_suffix` is :const:`None`, then locks the
        factory and creates stages with a temporary (UUID) value.  If
        the configuration has `cleanup_tmp_files` set to :const:`True`
        (the default) then executing the resulting pipeline will clean
        up the directory afterwards.

        :param dict config: `streamcorpus_pipeline` configuration
        :return: new pipeline instance