import functools


def coroutine(fn):
    def wrapper(*args, **kwargs):
        generator = fn(*args, **kwargs)
        next(generator)
        return generator

    return wrapper


class CollectionPipelineProcessor:
    sink = None
    start_source = None
    receiver = None

    def process(self, item):
        raise NotImplementedError

    def on_done(self):
        if self.receiver:
            self.receiver.close()

    def source(self, start_source):
        self.start_source = start_source

    @coroutine
    def make_generator(self):
        while True:
            try:
                item = yield
                self.process(item)
            except GeneratorExit:
                self.on_done()
                break

    def __or__(self, other):
        self.sink = other

        def exec():
            self.receiver = self.sink.make_generator()
            self.start_source()
        other.source(exec)

        return other

import functools


def coroutine(fn):
    def wrapper(*args, **kwargs):
        generator = fn(*args, **kwargs)
        next(generator)
        return generator

    return wrapper


class CollectionPipelineProcessor:
    sink = None
    start_source = None
    receiver = None

    def process(self, item):
        raise NotImplementedError

    def on_done(self):
        if self.receiver:
            self.receiver.close()

    def source(self, start_source):
        self.start_source = start_source

    @coroutine
    def make_generator(self):
        while True:
            try:
                item = yield
                self.process(item)
            except GeneratorExit:
                self.on_done()
                break

    def __or__(self, other):
        self.sink = other

        def exec():
            self.receiver = self.sink.make_generator()
            self.start_source()
        other.source(exec)

        return other


class CollectionPipelineOutput(CollectionPipelineProcessor):
    """Pipeline processor that ends the chain and starts outputing stream.

    Output processor immediately starts consuming from the source.
    Thus triggering the whole pipeline start.
    """
    def source(self, start_source):
        start_source()

Add base class for output pipeline processors
Add base class for output pipeline processors

  import functools
  
  
  def coroutine(fn):
      def wrapper(*args, **kwargs):
          generator = fn(*args, **kwargs)
          next(generator)
          return generator
  
      return wrapper
  
  
  class CollectionPipelineProcessor:
      sink = None
      start_source = None
      receiver = None
  
      def process(self, item):
          raise NotImplementedError
  
      def on_done(self):
          if self.receiver:
              self.receiver.close()
  
      def source(self, start_source):
          self.start_source = start_source
  
      @coroutine
      def make_generator(self):
          while True:
              try:
                  item = yield
                  self.process(item)
              except GeneratorExit:
                  self.on_done()
                  break
  
      def __or__(self, other):
          self.sink = other
  
          def exec():
              self.receiver = self.sink.make_generator()
              self.start_source()
          other.source(exec)
  
          return other
  
+ 
+ class CollectionPipelineOutput(CollectionPipelineProcessor):
+     """Pipeline processor that ends the chain and starts outputing stream.
+ 
+     Output processor immediately starts consuming from the source.
+     Thus triggering the whole pipeline start.
+     """
+     def source(self, start_source):
+         start_source()
+ 
Add base class for output pipeline processors
## Code Before:
import functools


def coroutine(fn):
    def wrapper(*args, **kwargs):
        generator = fn(*args, **kwargs)
        next(generator)
        return generator

    return wrapper


class CollectionPipelineProcessor:
    sink = None
    start_source = None
    receiver = None

    def process(self, item):
        raise NotImplementedError

    def on_done(self):
        if self.receiver:
            self.receiver.close()

    def source(self, start_source):
        self.start_source = start_source

    @coroutine
    def make_generator(self):
        while True:
            try:
                item = yield
                self.process(item)
            except GeneratorExit:
                self.on_done()
                break

    def __or__(self, other):
        self.sink = other

        def exec():
            self.receiver = self.sink.make_generator()
            self.start_source()
        other.source(exec)

        return other

## Instruction:
Add base class for output pipeline processors
## Code After:
import functools


def coroutine(fn):
    def wrapper(*args, **kwargs):
        generator = fn(*args, **kwargs)
        next(generator)
        return generator

    return wrapper


class CollectionPipelineProcessor:
    sink = None
    start_source = None
    receiver = None

    def process(self, item):
        raise NotImplementedError

    def on_done(self):
        if self.receiver:
            self.receiver.close()

    def source(self, start_source):
        self.start_source = start_source

    @coroutine
    def make_generator(self):
        while True:
            try:
                item = yield
                self.process(item)
            except GeneratorExit:
                self.on_done()
                break

    def __or__(self, other):
        self.sink = other

        def exec():
            self.receiver = self.sink.make_generator()
            self.start_source()
        other.source(exec)

        return other


class CollectionPipelineOutput(CollectionPipelineProcessor):
    """Pipeline processor that ends the chain and starts outputing stream.

    Output processor immediately starts consuming from the source.
    Thus triggering the whole pipeline start.
    """
    def source(self, start_source):
        start_source()
