def stream(self):
iterator = iter(self._stream)<EOL>while self._state < Computation.STATE_COMPLETED:<EOL><INDENT>try:<EOL><INDENT>message = next(iterator)<EOL><DEDENT>except StopIteration:<EOL><INDENT>if self._state < Computation.STATE_COMPLETED:<EOL><INDENT>self._stream = self._execute()<EOL>iterator = iter(self._stream)<EOL>continue<EOL><DEDENT><DEDENT>if isinstance(message, messages.StreamStartMessage):<EOL><INDENT>self._state = Computation.STATE_STREAM_STARTED<EOL>continue<EOL><DEDENT>if isinstance(message, messages.JobStartMessage):<EOL><INDENT>self._state = Computation.STATE_COMPUTATION_STARTED<EOL>self._id = message.handle<EOL>yield message<EOL>continue<EOL><DEDENT>if isinstance(message, messages.JobProgressMessage):<EOL><INDENT>yield message<EOL>continue<EOL><DEDENT>if isinstance(message, messages.ChannelAbortMessage):<EOL><INDENT>self._state = Computation.STATE_ABORTED<EOL>raise errors.ComputationAborted(message.abort_info)<EOL><DEDENT>if isinstance(message, messages.EndOfChannelMessage):<EOL><INDENT>self._state = Computation.STATE_COMPLETED<EOL>continue<EOL><DEDENT>if isinstance(message, messages.MetadataMessage):<EOL><INDENT>self._metadata[message.tsid] = message.properties<EOL>yield message<EOL>continue<EOL><DEDENT>if isinstance(message, messages.ExpiredTsIdMessage):<EOL><INDENT>if message.tsid in self._metadata:<EOL><INDENT>del self._metadata[message.tsid]<EOL><DEDENT>yield message<EOL>continue<EOL><DEDENT>if isinstance(message, messages.InfoMessage):<EOL><INDENT>self._process_info_message(message.message)<EOL>self._batch_count_detected = True<EOL>if self._current_batch_message:<EOL><INDENT>yield self._get_batch_to_yield()<EOL><DEDENT>continue<EOL><DEDENT>if isinstance(message, messages.DataMessage):<EOL><INDENT>self._state = Computation.STATE_DATA_RECEIVED<EOL>if not self._batch_count_detected:<EOL><INDENT>self._expected_batches += <NUM_LIT:1><EOL><DEDENT>if not self._current_batch_message:<EOL><INDENT>self._current_batch_message = message<EOL>self._current_batch_count = <NUM_LIT:1><EOL><DEDENT>elif (message.logical_timestamp_ms ==<EOL>self._current_batch_message.logical_timestamp_ms):<EOL><INDENT>self._current_batch_message.add_data(message.data)<EOL>self._current_batch_count += <NUM_LIT:1><EOL><DEDENT>else:<EOL><INDENT>self._batch_count_detected = True<EOL><DEDENT>if (self._batch_count_detected and<EOL>self._current_batch_count == self._expected_batches):<EOL><INDENT>yield self._get_batch_to_yield()<EOL><DEDENT>continue<EOL><DEDENT>if isinstance(message, messages.EventMessage):<EOL><INDENT>yield message<EOL>continue<EOL><DEDENT>if isinstance(message, messages.ErrorMessage):<EOL><INDENT>raise errors.ComputationFailed(message.errors)<EOL><DEDENT><DEDENT>if self._current_batch_message:<EOL><INDENT>yield self._get_batch_to_yield()<EOL><DEDENT>
Iterate over the messages from the computation's output.

        Control and metadata messages are intercepted and interpreted to
        enhance this Computation's object knowledge of the computation's
        context. Data and event messages are yielded back to the caller as a
        generator.