docs: def stream(self):
          """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.
          """
  
          iterator = iter(self._stream)
          while self._state < Computation.STATE_COMPLETED:
              try:
                  message = next(iterator)
              except StopIteration:
                  if self._state < Computation.STATE_COMPLETED:
                      self._stream = self._execute()
                      iterator = iter(self._stream)
                      continue
  
              if isinstance(message, messages.StreamStartMessage):
                  self._state = Computation.STATE_STREAM_STARTED
                  continue
  
              if isinstance(message, messages.JobStartMessage):
                  self._state = Computation.STATE_COMPUTATION_STARTED
                  self._id = message.handle
                  yield message
                  continue
  
              if isinstance(message, messages.JobProgressMessage):
                  yield message
                  continue
  
              if isinstance(message, messages.ChannelAbortMessage):
                  self._state = Computation.STATE_ABORTED
                  raise errors.ComputationAborted(message.abort_info)
  
              if isinstance(message, messages.EndOfChannelMessage):
                  self._state = Computation.STATE_COMPLETED
                  continue
  
              # Intercept metadata messages to accumulate received metadata...
              if isinstance(message, messages.MetadataMessage):
                  self._metadata[message.tsid] = message.properties
                  yield message
                  continue
  
              # ...as well as expired-tsid messages to clean it up.
              if isinstance(message, messages.ExpiredTsIdMessage):
                  if message.tsid in self._metadata:
                      del self._metadata[message.tsid]
                  yield message
                  continue
  
              if isinstance(message, messages.InfoMessage):
                  self._process_info_message(message.message)
                  self._batch_count_detected = True
                  if self._current_batch_message:
                      yield self._get_batch_to_yield()
                  continue
  
              # Accumulate data messages and release them when we have received
              # all batches for the same logical timestamp.
              if isinstance(message, messages.DataMessage):
                  self._state = Computation.STATE_DATA_RECEIVED
  
                  if not self._batch_count_detected:
                      self._expected_batches += 1
  
                  if not self._current_batch_message:
                      self._current_batch_message = message
                      self._current_batch_count = 1
                  elif (message.logical_timestamp_ms ==
                          self._current_batch_message.logical_timestamp_ms):
                      self._current_batch_message.add_data(message.data)
                      self._current_batch_count += 1
                  else:
                      self._batch_count_detected = True
  
                  if (self._batch_count_detected and
                          self._current_batch_count == self._expected_batches):
                      yield self._get_batch_to_yield()
                  continue
  
              if isinstance(message, messages.EventMessage):
                  yield message
                  continue
  
              if isinstance(message, messages.ErrorMessage):
                  raise errors.ComputationFailed(message.errors)
  
          # Yield last batch, even if potentially incomplete.
          if self._current_batch_message:
              yield self._get_batch_to_yield()