code: def _enqueue_batches(self, stop_event: Event) -> None:
          
          while True:
              self._stream = self._get_stream()
              while True:
                  # Acquire the semaphore before processing the next batch
                  # but immediately release it so that other threads
                  # are not blocked when they decide to acquire it again.
                  with self._semaphore:
                      pass
                  # It always takes a short moment before the native call actually
                  # releases the GIL and we are free to compute. The following sleep
                  # is here to compensate for this short moment - we don't want to 
                  # slow down the native call before the GIL is released.
                  time.sleep(CXF_BUFFER_SLEEP)
                  try:
                      batch = next(self._stream)
                  except StopIteration:
                      break
                  self._queue.put(batch)
                  self._batch_count += 1
                  if stop_event.is_set():
                      return
                  if self._epoch_limit_reached():
                      self._queue.put(None)
                      self._batch_count = 0
                      return
              self._stream = None  # yield a new iterator next time
              if self._epoch_size <= 0:  # for non-fixed size epochs
                  self._queue.put(None)
                  self._batch_count = 0
                  return
docs: Enqueue all the stream batches. If specified, stop after ``epoch_size`` batches.
  
          .. note::
              Signal the epoch end with ``None``.
  
          Stop when:
          - ``stop_event`` is risen
          - stream ends and epoch size is not set
          - specified number of batches is enqueued
  
          .. note::
              This is used only with ``buffer`` > 0.
  
          :param stop_event: event signaling stop instruction