m.sendMu.Lock()
	defer m.sendMu.Unlock()
	msg.MuxID = m.ID
	msg.Seq = m.SendSeq
	m.SendSeq++
	if debugPrint {
		fmt.Printf("Mux %d, Sending %+v\n", m.ID, msg)
	}
	gridLogIf(m.ctx, m.parent.queueMsg(msg, nil))
}

func (m *muxServer) close() {
	m.cancel()
	m.recvMu.Lock()
	defer m.recvMu.Unlock()

	if m.inbound != nil {
		xioutil.SafeClose(m.inbound)
		m.inbound = nil
	}

 *
 * Work within queues is not concurrent. This is equivalent to each queue having a dedicated thread
 * for its work; in practice a set of queues may share a set of threads to save resources.
 */
class TaskQueue internal constructor(
  internal val taskRunner: TaskRunner,
  internal val name: String,
) {
  internal var shutdown = false

    //    second thread can then dispatch to the subscriber it took before the first thread does.
    //
    // All this makes me really wonder if there's any value in queueing here at all. A dispatcher
    // that simply loops through the subscribers and dispatches the event to each would actually
    // probably provide a stronger order guarantee, though that order would obviously be different

     * @param sessionId The session ID.
     */
    void delete(String sessionId);

    /**
     * Deletes all URL queues.
     */
    void deleteAll();

    /**
     * Offers all URL queues.
     *
     * @param sessionId The session ID.
     * @param newUrlQueueList The list of new URL queues.
     */
    void offerAll(String sessionId, List<QUEUE> newUrlQueueList);

    /**
     * Polls a URL queue.
     *

 *
 * <p>
 * The class is responsible for:
 * </p>
 * <ul>
 *   <li>Updating session IDs for URL queues.</li>
 *   <li>Adding new URLs to the queue.</li>
 *   <li>Inserting existing {@link UrlQueueImpl} objects into the queue.</li>
 *   <li>Deleting URL queues associated with a session.</li>
 *   <li>Deleting all URL queues.</li>
 *   <li>Offering a list of URLs to the queue, ensuring duplicates are not added.</li>

 *
 * <p>The class also provides methods to manage URL patterns for inclusion and exclusion, which
 * are stored as {@code Pattern} objects.
 */
public class MemoryDataHelper {
    /** Map of session IDs to URL queues for managing crawling queues. */
    protected volatile Map<String, Queue<UrlQueueImpl<Long>>> urlQueueMap = new HashMap<>();

    /** Map of session IDs to access result maps for storing crawling results. */

 * decide not to by returning -1L.
 *
 * Task Queues
 * -----------
 *
 * Tasks are bound to the [TaskQueue] they are scheduled in. Each queue is sequential and the tasks
 * within it never execute concurrently. It is an error to use a task in multiple queues.
 */
abstract class Task(
  val name: String,
  val cancelable: Boolean = true,
) {

package com.google.common.collect.testing;

import com.google.common.annotations.GwtCompatible;
import java.util.Queue;
import org.jspecify.annotations.NullMarked;
import org.jspecify.annotations.Nullable;

/**
 * Creates queues, containing sample elements, to be tested.
 *
 * @author Jared Levy
 */
@GwtCompatible
@NullMarked
public interface TestQueueGenerator<E extends @Nullable Object> extends TestCollectionGenerator<E> {
  @Override

package com.google.common.collect.testing;

import com.google.common.annotations.GwtCompatible;
import java.util.Queue;
import org.jspecify.annotations.NullMarked;
import org.jspecify.annotations.Nullable;

/**
 * Creates queues, containing sample elements, to be tested.
 *
 * @author Jared Levy
 */
@GwtCompatible
@NullMarked
public interface TestQueueGenerator<E extends @Nullable Object> extends TestCollectionGenerator<E> {
  @Override

import java.util.LinkedList;
import junit.framework.TestCase;
import org.jspecify.annotations.NullUnmarked;
import org.jspecify.annotations.Nullable;

/**
 * Tests for {@link Synchronized#deque} and {@link Queues#synchronizedDeque}.
 *
 * @author Kurt Alfred Kluever
 */
@NullUnmarked
public class SynchronizedDequeTest extends TestCase {

  protected Deque<String> create() {
    TestDeque<String> inner = new TestDeque<>();