private val messageAndCloseQueue = ArrayDeque<Any>()

  /** The total size in bytes of enqueued but not yet transmitted messages. */
  private var queueSize = 0L

  /** True if we've enqueued a close frame. No further message frames will be enqueued. */
  private var enqueuedClose = false

  /** The close code from the peer, or -1 if this web socket has not yet read a close frame. */

	if err != nil {
		t.Fatalf("error creating Deployment controller: %v", err)
	}
	enqueued := false
	c.enqueueDeployment = func(d *apps.Deployment) {
		if d.Name == "foo" {
			enqueued = true
		}
	}

	c.deletePod(logger, pod)

	if !enqueued {
		t.Errorf("expected deployment %q to be queued after pod deletion", foo.Name)
	}
}

	}()

	// The controller is blocked because the reconcile function sends on an
	// unbuffered channel.
	require.False(t, myController.HasSynced())

	// Wait for all enqueued reconciliations
	require.NoError(t, waitForReconcile(initialObject))

	// Now it is safe to wait for it to Sync
	require.True(t, cache.WaitForCacheSync(testContext.Done(), myController.HasSynced))

  }

  private enum class ResumePriority {
    /** Resumes as soon as the condition is satisfied. */
    BeforeOtherTasks,

    /** Resumes after the already-enqueued tasks. */
    AfterEnqueuedTasks,

    /** Resumes after all other tasks, including tasks enqueued while yielding. */
    AfterOtherTasks,
  }

  /** Returns the task that was started, or null if there were no tasks to start. */

  std::unique_ptr<AsyncValueAllocator> pjrt_allocator_ TF_GUARDED_BY(mu_);

  // Stream associated with this device. Operations enqueued on this
  // stream are executed on the device. Operations include data
  // copying back and forth between CPU and the device, and
  // computations enqueued by XLA.
  std::shared_ptr<se::Stream> stream_ TF_GUARDED_BY(mu_);
  // If false, only stream_ is valid and all computation and transfers use

	if key == "" || done {
		t.Error("failed to enqueue StatefulSet")
	} else if expectedKey, _ := controller.KeyFunc(set1); expectedKey != key {
		t.Errorf("expected StatefulSet key %s found %s", expectedKey, key)
	}
	ssc.queue.Done(key)

	ssc.addPod(logger, pod2)
	key, done = ssc.queue.Get()
	if key == "" || done {
		t.Error("failed to enqueue StatefulSet")

      "FINE: Q10000 scheduled after 100 µs: task",
      "FINE: Q10000 starting              : task",
      "FINE: Q10000 finished run in   0 µs: task",
    )
  }

  /** Inspect how many runnables have been enqueued. If none then we're truly sequential. */
  @Test fun singleQueueIsSerial() {
    redQueue.execute("task one", 100.µs) {
      log += "one:run@${taskFaker.nanoTime} parallel=${taskFaker.isParallel}"
    }

   * Scripts [response] to be returned to a request made in sequence. The first request is
   * served by the first enqueued response; the second request by the second enqueued response; and
   * so on.
   *
   * @throws ClassCastException if the default dispatcher has been
   * replaced with [setDispatcher][dispatcher].
   */
  fun enqueue(response: MockResponse) = (dispatcher as QueueDispatcher).enqueueResponse(response)

  /**

    }

    /**
     * Returns true if the request contains conditions that save the server from sending a response
     * that the client has locally. When a request is enqueued with its own conditions, the built-in
     * response cache won't be used.
     */
    private fun hasConditions(request: Request): Boolean =

	New: func() any { return new(packet) },
}

type packetQueueState struct {
	head, tail      *packet // unqueued packets
	nBytes          int     // number of bytes enqueued in the packet buffers starting from head
	readBufferBytes int     // soft limit on nbytes; no more packets may be enqueued when the limit is exceeded
	readClosed      bool    // true if the reader of the queue has stopped reading