Connects the ``Piper`` instance to its upstream ``Pipers`` that should
        be given as a sequence. This connects this ``Piper.inbox`` with the
        upstream ``Piper.outbox`` respecting any "consume", "spawn" and 
        "produce" arguments. 
        
        Arguments:
        
          - inbox(sequence) sequence of ``Piper`` instances.

def connect(self, inbox):
        """
        Connects the ``Piper`` instance to its upstream ``Pipers`` that should
        be given as a sequence. This connects this ``Piper.inbox`` with the
        upstream ``Piper.outbox`` respecting any "consume", "spawn" and 
        "produce" arguments. 
        
        Arguments:
        
          - inbox(sequence) sequence of ``Piper`` instances.
        
        """
        if self.started:
            self.log.error('Piper %s is started and cannot connect to %s.' % \
                           (self, inbox))
            raise PiperError('Piper %s is started and cannot connect to %s.' % \
                             (self, inbox))
        elif self.connected:
            self.log.error('Piper %s is connected and cannot connect to %s.' % \
                           (self, inbox))
            raise PiperError('Piper %s is connected and cannot connect to %s.' % \
                             (self, inbox))
        elif hasattr(self.imap, '_started') and self.imap._started.isSet():
            self.log.error('Piper %s cannot connect (NuMap is started).' % \
                           self)
            raise PiperError('Piper %s cannot connect (NuMap is started).' % \
                           self)
        else:
            # not started and not connected and NuMap not started
            self.log.debug('Piper %s connects to %s' % (self, inbox))
            # determine the stride with which result will be consumed from the
            # input.
            stride = self.imap.stride if hasattr(self.imap, 'stride') else 1

            # Tee input iterators. The idea is to create a promise object for a
            # tee. The actual teed iterator will be created on start. Each tee
            # is protected with a seperate lock the reasons for this are:
            # - tee objects are as a collection not thread safe
            # - tee objects might be next'ed from different threads, a single
            #   lock will not guarantee that a thread might be allowed to finish
            #   it's stride. (How it works that a thread releases the next 
            #   thread only if it finished a stride
            teed = []
            for piper in inbox:
                if hasattr(piper, '_iter'): # isinstance Piper?
                    piper.tee_num += 1
                    tee_lock = Lock()
                    tee_lock.acquire()
                    piper.tee_locks.append(tee_lock)
                    piper = _TeePiper(piper, piper.tee_num - 1, stride)
                teed.append(_InputIterator(piper, self))

            # set how much to consume from input iterators.
            self.inbox = _Zip(*teed) if self.consume == 1 else\
                 _Consume(_Zip(*teed), n=self.consume, stride=stride)

            # set how much to 
            for i in xrange(self.spawn):
                self.imap_tasks.append(\
                    self.imap(self.worker, self.inbox) \
                        if self.imap is imap else \
                    self.imap(self.worker, self.inbox, timeout=self.timeout, \
                              track=self.track))
            # chain the results together.
            outbox = _Chain(self.imap_tasks, stride=stride)
            # Make output
            #prd = ProduceFromSequence if self.produce_from_sequence else Produce
            if self.produce == 1:
                self.outbox = outbox
            elif self.repeat:
                self.outbox = _Repeat(outbox, n=self.produce, stride=stride)
            else:
                self.outbox = _Produce(outbox, n=self.produce, stride=stride)
            self.connected = True

        return self