def connect(self, inbox):
if self.started:<EOL><INDENT>self.log.error('<STR_LIT>' %(self, inbox))<EOL>raise PiperError('<STR_LIT>' %(self, inbox))<EOL><DEDENT>elif self.connected:<EOL><INDENT>self.log.error('<STR_LIT>' %(self, inbox))<EOL>raise PiperError('<STR_LIT>' %(self, inbox))<EOL><DEDENT>elif hasattr(self.imap, '<STR_LIT>') and self.imap._started.isSet():<EOL><INDENT>self.log.error('<STR_LIT>' %self)<EOL>raise PiperError('<STR_LIT>' %self)<EOL><DEDENT>else:<EOL><INDENT>self.log.debug('<STR_LIT>' % (self, inbox))<EOL>stride = self.imap.stride if hasattr(self.imap, '<STR_LIT>') else <NUM_LIT:1><EOL>teed = []<EOL>for piper in inbox:<EOL><INDENT>if hasattr(piper, '<STR_LIT>'): <EOL><INDENT>piper.tee_num += <NUM_LIT:1><EOL>tee_lock = Lock()<EOL>tee_lock.acquire()<EOL>piper.tee_locks.append(tee_lock)<EOL>piper = _TeePiper(piper, piper.tee_num - <NUM_LIT:1>, stride)<EOL><DEDENT>teed.append(_InputIterator(piper, self))<EOL><DEDENT>self.inbox = _Zip(*teed) if self.consume == <NUM_LIT:1> else_Consume(_Zip(*teed), n=self.consume, stride=stride)<EOL>for i in range(self.spawn):<EOL><INDENT>self.imap_tasks.append(self.imap(self.worker, self.inbox)if self.imap is imap elseself.imap(self.worker, self.inbox, timeout=self.timeout,track=self.track))<EOL><DEDENT>outbox = _Chain(self.imap_tasks, stride=stride)<EOL>if self.produce == <NUM_LIT:1>:<EOL><INDENT>self.outbox = outbox<EOL><DEDENT>elif self.repeat:<EOL><INDENT>self.outbox = _Repeat(outbox, n=self.produce, stride=stride)<EOL><DEDENT>else:<EOL><INDENT>self.outbox = _Produce(outbox, n=self.produce, stride=stride)<EOL><DEDENT>self.connected = True<EOL><DEDENT>return self<EOL>
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.