Java 类java.util.concurrent.Exchanger 实例源码

@Override
public RpiUnit init(Object input) {
    if (Objects.nonNull(executorForAgents)) {
        this.agents = new ArrayList<>();
        this.active = new AtomicBoolean(false);
        this.commandQueue = new LinkedBlockingQueue<>();
        SimpleLoggingUtil.print(getClass(), "TankRpi: INIT");
        final Exchanger<GenericCommand<PlatformUnitCommandEnum>> platformExchanger = new Exchanger<>();

        final Map<String, GenericMotor> enginesMap = EngineRegistry.getInstance().getByNames(CONSUMER_NAME);

        this.agents.add(createAgent("platformAgent", new ClientPlatformProducer(commandQueue, platformExchanger),
                new ClientPlatformConsumer(executorForAgents, platformExchanger, enginesMap)));

        if (!agents.isEmpty()) {
            active.set(true);
            logic = initLogic();
        }
    }

    return this;
}

public static void main(String[] args) {

        // 创建两个缓冲区
        List<String> buffer1=new ArrayList<>();
        List<String> buffer2=new ArrayList<>();

        // 创建 Exchanger
        Exchanger<List<String>> exchanger=new Exchanger<>();

        // 创建生产者
        Producer producer=new Producer(buffer1, exchanger);
        // 创建消费者
        Consumer consumer=new Consumer(buffer2, exchanger);

        Thread threadProducer=new Thread(producer);
        Thread threadConsumer=new Thread(consumer);

        threadProducer.start();
        threadConsumer.start();
    }

/**
 * Main method of the example
 * @param args
 */
public static void main(String[] args) {

    // Creates two buffers
    List<String> buffer1=new ArrayList<>();
    List<String> buffer2=new ArrayList<>();

    // Creates the exchanger
    Exchanger<List<String>> exchanger=new Exchanger<>();

    // Creates the producer
    Producer producer=new Producer(buffer1, exchanger);
    // Creates the consumer
    Consumer consumer=new Consumer(buffer2, exchanger);

    // Creates and starts the threads
    Thread threadProducer=new Thread(producer);
    Thread threadConsumer=new Thread(consumer);

    threadProducer.start();
    threadConsumer.start();

}

@Override
protected void clientClosed(RMIConnection conn) throws IOException {
    System.out.println("clientClosed, will call connectorServer.stop");
    final Exchanger<Void> x = new Exchanger<Void>();
    Thread t = new Thread() {
        public void run() {
            try {
                connectorServer.stop();
            } catch (Exception e) {
                fail(e);
            }
        }
    };
    t.setName("connectorServer.stop");
    t.start();
    waitForBlock(t);
    /* If this thread is synchronized on RMIServerImpl, then
     * the thread that does connectorServer.stop will acquire
     * the clientList lock and then block waiting for the RMIServerImpl
     * lock.  Our call to super.clientClosed will then deadlock because
     * it needs to acquire the clientList lock.
     */
    System.out.println("calling super.clientClosed");
    System.out.flush();
    super.clientClosed(conn);
}

/**
 * exchange exchanges objects across two threads
 */
public void testExchange() {
    final Exchanger e = new Exchanger();
    Thread t1 = newStartedThread(new CheckedRunnable() {
        public void realRun() throws InterruptedException {
            assertSame(one, e.exchange(two));
            assertSame(two, e.exchange(one));
        }});
    Thread t2 = newStartedThread(new CheckedRunnable() {
        public void realRun() throws InterruptedException {
            assertSame(two, e.exchange(one));
            assertSame(one, e.exchange(two));
        }});

    awaitTermination(t1);
    awaitTermination(t2);
}

/**
 * timed exchange exchanges objects across two threads
 */
public void testTimedExchange() {
    final Exchanger e = new Exchanger();
    Thread t1 = newStartedThread(new CheckedRunnable() {
        public void realRun() throws Exception {
            assertSame(one, e.exchange(two, LONG_DELAY_MS, MILLISECONDS));
            assertSame(two, e.exchange(one, LONG_DELAY_MS, MILLISECONDS));
        }});
    Thread t2 = newStartedThread(new CheckedRunnable() {
        public void realRun() throws Exception {
            assertSame(two, e.exchange(one, LONG_DELAY_MS, MILLISECONDS));
            assertSame(one, e.exchange(two, LONG_DELAY_MS, MILLISECONDS));
        }});

    awaitTermination(t1);
    awaitTermination(t2);
}

static void oneRun(int nthreads, int iters) throws Exception {
    LoopHelpers.BarrierTimer timer = new LoopHelpers.BarrierTimer();
    CyclicBarrier barrier = new CyclicBarrier(nthreads + 1, timer);
    Exchanger<Int> l = null;
    Exchanger<Int> r = new Exchanger<>();
    for (int i = 0; i < nthreads; ++i) {
        pool.execute(new Stage(l, r, barrier, iters));
        l = r;
        r = (i+2 < nthreads) ? new Exchanger<Int>() : null;
    }
    barrier.await();
    barrier.await();
    long time = timer.getTime();
    if (print)
        System.out.println(LoopHelpers.rightJustify(time / (iters * nthreads + iters * (nthreads-2))) + " ns per transfer");
}

@Override
protected void clientClosed(RMIConnection conn) throws IOException {
    System.out.println("clientClosed, will call connectorServer.stop");
    final Exchanger<Void> x = new Exchanger<Void>();
    Thread t = new Thread() {
        public void run() {
            try {
                connectorServer.stop();
            } catch (Exception e) {
                fail(e);
            }
        }
    };
    t.setName("connectorServer.stop");
    t.start();
    waitForBlock(t);
    /* If this thread is synchronized on RMIServerImpl, then
     * the thread that does connectorServer.stop will acquire
     * the clientList lock and then block waiting for the RMIServerImpl
     * lock.  Our call to super.clientClosed will then deadlock because
     * it needs to acquire the clientList lock.
     */
    System.out.println("calling super.clientClosed");
    System.out.flush();
    super.clientClosed(conn);
}

/**
 * Main method of the example
 * @param args
 */
public static void main(String[] args) {

    // Creates two buffers
    List<String> buffer1=new ArrayList<>();
    List<String> buffer2=new ArrayList<>();

    // Creates the exchanger
    Exchanger<List<String>> exchanger=new Exchanger<>();

    // Creates the producer
    Producer producer=new Producer(buffer1, exchanger);
    // Creates the consumer
    Consumer consumer=new Consumer(buffer2, exchanger);

    // Creates and starts the threads
    Thread threadProducer=new Thread(producer);
    Thread threadConsumer=new Thread(consumer);

    threadProducer.start();
    threadConsumer.start();

}

public static void main(String[] args) throws Exception
{
    if (args.length > 0)
    {
        size = new Integer(args[0]);
    }
    if (args.length > 1)
    {
        delay = new Integer(args[1]);
    }
    ExecutorService exec = Executors.newCachedThreadPool();
    Exchanger<List<Fat>> xc = new Exchanger<List<Fat>>();
    List<Fat> producerList = new CopyOnWriteArrayList<Fat>(), consumerList = new CopyOnWriteArrayList<Fat>();
    exec.execute(new ExchangerProducer<Fat>(xc, BasicGenerator
            .create(Fat.class), producerList));
    exec.execute(new ExchangerConsumer<Fat>(xc, consumerList));
    TimeUnit.SECONDS.sleep(delay);
    exec.shutdownNow();
}

/**
 * exchange exchanges objects across two threads
 */
public void testExchange() {
    final Exchanger e = new Exchanger();
    Thread t1 = newStartedThread(new CheckedRunnable() {
        public void realRun() throws InterruptedException {
            assertSame(one, e.exchange(two));
            assertSame(two, e.exchange(one));
        }});
    Thread t2 = newStartedThread(new CheckedRunnable() {
        public void realRun() throws InterruptedException {
            assertSame(two, e.exchange(one));
            assertSame(one, e.exchange(two));
        }});

    awaitTermination(t1);
    awaitTermination(t2);
}

/**
 * timed exchange exchanges objects across two threads
 */
public void testTimedExchange() {
    final Exchanger e = new Exchanger();
    Thread t1 = newStartedThread(new CheckedRunnable() {
        public void realRun() throws Exception {
            assertSame(one, e.exchange(two, LONG_DELAY_MS, MILLISECONDS));
            assertSame(two, e.exchange(one, LONG_DELAY_MS, MILLISECONDS));
        }});
    Thread t2 = newStartedThread(new CheckedRunnable() {
        public void realRun() throws Exception {
            assertSame(two, e.exchange(one, LONG_DELAY_MS, MILLISECONDS));
            assertSame(one, e.exchange(two, LONG_DELAY_MS, MILLISECONDS));
        }});

    awaitTermination(t1);
    awaitTermination(t2);
}

static void oneRun(int nthreads, int iters) throws Exception {
    LoopHelpers.BarrierTimer timer = new LoopHelpers.BarrierTimer();
    CyclicBarrier barrier = new CyclicBarrier(nthreads + 1, timer);
    Exchanger<Int> l = null;
    Exchanger<Int> r = new Exchanger<Int>();
    for (int i = 0; i < nthreads; ++i) {
        pool.execute(new Stage(l, r, barrier, iters));
        l = r;
        r = (i+2 < nthreads) ? new Exchanger<Int>() : null;
    }
    barrier.await();
    barrier.await();
    long time = timer.getTime();
    if (print)
        System.out.println(LoopHelpers.rightJustify(time / (iters * nthreads + iters * (nthreads-2))) + " ns per transfer");
}

@Override
protected void clientClosed(RMIConnection conn) throws IOException {
    System.out.println("clientClosed, will call connectorServer.stop");
    final Exchanger<Void> x = new Exchanger<Void>();
    Thread t = new Thread() {
        public void run() {
            try {
                connectorServer.stop();
            } catch (Exception e) {
                fail(e);
            }
        }
    };
    t.setName("connectorServer.stop");
    t.start();
    waitForBlock(t);
    /* If this thread is synchronized on RMIServerImpl, then
     * the thread that does connectorServer.stop will acquire
     * the clientList lock and then block waiting for the RMIServerImpl
     * lock.  Our call to super.clientClosed will then deadlock because
     * it needs to acquire the clientList lock.
     */
    System.out.println("calling super.clientClosed");
    System.out.flush();
    super.clientClosed(conn);
}

public static void main(String []args) {
    final Exchanger <Integer>exchanger = new Exchanger<Integer>();
    for(int i = 0 ; i < 10 ; i++) {
        final Integer num = i;
        new Thread() {
            public void run() {
                System.out.println("我是线程：Thread_" + this.getName() + " 我的数据是：" + num);
                try {
                    Integer exchangeNum = exchanger.exchange(num);
                    Thread.sleep(1000);
                    System.out.println("我是线程：Thread_" + this.getName() + " 最初的数据为：" + num + " , 交换后的数据为：" + exchangeNum);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }.start();
    }
}

/**
 * http://ifeve.com/thread-synchronization-utilities-8/
 * 
 * @see ExchangerProducer
 * @see ExchangerConsumer
 */
public static void producerAndConsumerAndExchangerMain() {
    List<String> buffer1 = new ArrayList<>();
    List<String> buffer2 = new ArrayList<>();

    Exchanger<List<String>> exchanger = new Exchanger<>();

    ExchangerProducer producer = new ExchangerProducer(buffer1, exchanger);
    ExchangerConsumer consumer = new ExchangerConsumer(buffer2, exchanger);

    Thread producerThread = new Thread(producer);
    Thread consumerThread = new Thread(consumer);

    producerThread.start();
    consumerThread.start();
}

public void testContendedPut() throws Exception
{
    final int max = 1000000;
    CachedQuery orderedData[] = new CachedQuery[max];
    for(int i=0;i<max;i++)
    {
        CachedQuery o = createCachedQuery(i);
        orderedData[i] = o;
    }
    final CachedQuery data[] = shuffle(orderedData);
    final Exchanger exchanger = new Exchanger();
    final NonLruQueryIndex index = new NonLruQueryIndex();
    PutRunnableWithExchange first = new PutRunnableWithExchange(0, max, data, index, exchanger);
    PutRunnableWithExchange second = new PutRunnableWithExchange(0, max, data, index, exchanger);
    ExceptionCatchingThread firstThread = new ExceptionCatchingThread(first);
    ExceptionCatchingThread secondThread = new ExceptionCatchingThread(second);
    firstThread.start();
    secondThread.start();
    firstThread.joinWithExceptionHandling();
    secondThread.joinWithExceptionHandling();
    assertEquals(max, index.size());
    assertEquals(max, index.getEntryCount());
    first.verifyExistence();
    second.verifyExistence();
}

public void testContendedGetIfAbsentPut() throws Exception
{
    final int max = 1000000;
    String orderedData[] = new String[max];
    for(int i=0;i<max;i++)
    {
        String o = createData(i);
        orderedData[i] = o;
    }
    final String data[] = shuffle(orderedData);
    final Exchanger exchanger = new Exchanger();
    final StringIndex index = createStringPool();
    GetIfAbsentPutRunnableWithExchange first = new GetIfAbsentPutRunnableWithExchange(0, max, data, index, exchanger);
    GetIfAbsentPutRunnableWithExchange second = new GetIfAbsentPutRunnableWithExchange(0, max, data, index, exchanger);
    ExceptionCatchingThread firstThread = new ExceptionCatchingThread(first);
    ExceptionCatchingThread secondThread = new ExceptionCatchingThread(second);
    firstThread.start();
    secondThread.start();
    firstThread.joinWithExceptionHandling();
    secondThread.joinWithExceptionHandling();
    assertEquals(max, index.size());
    first.verifyExistence();
    second.verifyExistence();
}

public void testContendedGetIfAbsentPut() throws Exception
{
    final int max = 1000000;
    String orderedData[] = new String[max];
    for(int i=0;i<max;i++)
    {
        String o = createData(i);
        orderedData[i] = o;
    }
    final String data[] = shuffle(orderedData);
    final Exchanger exchanger = new Exchanger();
    final ConcurrentWeakPool index = createStringPool();
    GetIfAbsentPutRunnableWithExchange first = new GetIfAbsentPutRunnableWithExchange(0, max, data, index, exchanger);
    GetIfAbsentPutRunnableWithExchange second = new GetIfAbsentPutRunnableWithExchange(0, max, data, index, exchanger);
    ExceptionCatchingThread firstThread = new ExceptionCatchingThread(first);
    ExceptionCatchingThread secondThread = new ExceptionCatchingThread(second);
    firstThread.start();
    secondThread.start();
    firstThread.joinWithExceptionHandling();
    secondThread.joinWithExceptionHandling();
    assertEquals(max, index.size());
    first.verifyExistence();
    second.verifyExistence();
}

public void testContendedGetIfAbsentPut() throws Exception
{
    final int max = 1000000;
    TestObject orderedData[] = new TestObject[max];
    for(int i=0;i<max;i++)
    {
        TestObject o = new TestObject(i);
        orderedData[i] = o;
    }
    final TestObject data[] = shuffle(orderedData);
    final Exchanger exchanger = new Exchanger();
    Extractor[] extractors = {new ShiftedIntExtractor(1)};
    final ConcurrentFullUniqueIndex<TestObject> index = new ConcurrentFullUniqueIndex(extractors, 7);
    PutIfAbsentRunnableWithExchange first = new PutIfAbsentRunnableWithExchange(0, max, data, index, exchanger);
    PutIfAbsentRunnableWithExchange second = new PutIfAbsentRunnableWithExchange(0, max, data, index, exchanger);
    ExceptionCatchingThread firstThread = new ExceptionCatchingThread(first);
    ExceptionCatchingThread secondThread = new ExceptionCatchingThread(second);
    firstThread.start();
    secondThread.start();
    firstThread.joinWithExceptionHandling();
    secondThread.joinWithExceptionHandling();
    assertEquals(max, index.size());
    first.verifyExistence();
    second.verifyExistence();
}

@Override
protected void clientClosed(RMIConnection conn) throws IOException {
    System.out.println("clientClosed, will call connectorServer.stop");
    final Exchanger<Void> x = new Exchanger<Void>();
    Thread t = new Thread() {
        public void run() {
            try {
                connectorServer.stop();
            } catch (Exception e) {
                fail(e);
            }
        }
    };
    t.setName("connectorServer.stop");
    t.start();
    waitForBlock(t);
    /* If this thread is synchronized on RMIServerImpl, then
     * the thread that does connectorServer.stop will acquire
     * the clientList lock and then block waiting for the RMIServerImpl
     * lock.  Our call to super.clientClosed will then deadlock because
     * it needs to acquire the clientList lock.
     */
    System.out.println("calling super.clientClosed");
    System.out.flush();
    super.clientClosed(conn);
}

@Override
protected void clientClosed(RMIConnection conn) throws IOException {
    System.out.println("clientClosed, will call connectorServer.stop");
    final Exchanger<Void> x = new Exchanger<Void>();
    Thread t = new Thread() {
        public void run() {
            try {
                connectorServer.stop();
            } catch (Exception e) {
                fail(e);
            }
        }
    };
    t.setName("connectorServer.stop");
    t.start();
    waitForBlock(t);
    /* If this thread is synchronized on RMIServerImpl, then
     * the thread that does connectorServer.stop will acquire
     * the clientList lock and then block waiting for the RMIServerImpl
     * lock.  Our call to super.clientClosed will then deadlock because
     * it needs to acquire the clientList lock.
     */
    System.out.println("calling super.clientClosed");
    System.out.flush();
    super.clientClosed(conn);
}

@Override
protected void clientClosed(RMIConnection conn) throws IOException {
    System.out.println("clientClosed, will call connectorServer.stop");
    final Exchanger<Void> x = new Exchanger<Void>();
    Thread t = new Thread() {
        public void run() {
            try {
                connectorServer.stop();
            } catch (Exception e) {
                fail(e);
            }
        }
    };
    t.setName("connectorServer.stop");
    t.start();
    waitForBlock(t);
    /* If this thread is synchronized on RMIServerImpl, then
     * the thread that does connectorServer.stop will acquire
     * the clientList lock and then block waiting for the RMIServerImpl
     * lock.  Our call to super.clientClosed will then deadlock because
     * it needs to acquire the clientList lock.
     */
    System.out.println("calling super.clientClosed");
    System.out.flush();
    super.clientClosed(conn);
}

@Override
protected void clientClosed(RMIConnection conn) throws IOException {
    System.out.println("clientClosed, will call connectorServer.stop");
    final Exchanger<Void> x = new Exchanger<Void>();
    Thread t = new Thread() {
        public void run() {
            try {
                connectorServer.stop();
            } catch (Exception e) {
                fail(e);
            }
        }
    };
    t.setName("connectorServer.stop");
    t.start();
    waitForBlock(t);
    /* If this thread is synchronized on RMIServerImpl, then
     * the thread that does connectorServer.stop will acquire
     * the clientList lock and then block waiting for the RMIServerImpl
     * lock.  Our call to super.clientClosed will then deadlock because
     * it needs to acquire the clientList lock.
     */
    System.out.println("calling super.clientClosed");
    System.out.flush();
    super.clientClosed(conn);
}

@Test
public void testRateLimit() throws Exception {
  final TestProducer p = new TestProducer();
  MockStages.setSourceCapture(p);

  final ProductionPipeline pipeline = createProductionPipeline(DeliveryGuarantee.AT_MOST_ONCE, true, 10L, PipelineType.DEFAULT);
  pipeline.registerStatusListener(new MyStateListener());
  final Exchanger<Double> rate = new Exchanger<>();
  new Thread() {
    @Override
    public void run() {
      try {
        long start = System.nanoTime();
        pipeline.run();
        rate.exchange(p.count.doubleValue() * 1000 * 1000 * 1000 / (System.nanoTime() - start));
      } catch (Exception ex) {

      }
    }
  }.start();
  Thread.sleep(10000);
  pipeline.stop();
  Double rateAchieved = rate.exchange(0.0);
  // To account for the slight loss of precision, we compare the "long-ified" versions.
  Assert.assertTrue(rateAchieved.longValue() <= 10);
}

@Override
protected void clientClosed(RMIConnection conn) throws IOException {
    System.out.println("clientClosed, will call connectorServer.stop");
    final Exchanger<Void> x = new Exchanger<Void>();
    Thread t = new Thread() {
        public void run() {
            try {
                connectorServer.stop();
            } catch (Exception e) {
                fail(e);
            }
        }
    };
    t.setName("connectorServer.stop");
    t.start();
    waitForBlock(t);
    /* If this thread is synchronized on RMIServerImpl, then
     * the thread that does connectorServer.stop will acquire
     * the clientList lock and then block waiting for the RMIServerImpl
     * lock.  Our call to super.clientClosed will then deadlock because
     * it needs to acquire the clientList lock.
     */
    System.out.println("calling super.clientClosed");
    System.out.flush();
    super.clientClosed(conn);
}

@Override
protected void clientClosed(RMIConnection conn) throws IOException {
    System.out.println("clientClosed, will call connectorServer.stop");
    final Exchanger<Void> x = new Exchanger<Void>();
    Thread t = new Thread() {
        public void run() {
            try {
                connectorServer.stop();
            } catch (Exception e) {
                fail(e);
            }
        }
    };
    t.setName("connectorServer.stop");
    t.start();
    waitForBlock(t);
    /* If this thread is synchronized on RMIServerImpl, then
     * the thread that does connectorServer.stop will acquire
     * the clientList lock and then block waiting for the RMIServerImpl
     * lock.  Our call to super.clientClosed will then deadlock because
     * it needs to acquire the clientList lock.
     */
    System.out.println("calling super.clientClosed");
    System.out.flush();
    super.clientClosed(conn);
}

@Test
public void testNotCloseZkWhenPending() throws Exception {
  ZooKeeper mockedZK = mock(ZooKeeper.class);
  Exchanger<AsyncCallback.DataCallback> exchanger = new Exchanger<>();
  doAnswer(i -> {
    exchanger.exchange(i.getArgument(2));
    return null;
  }).when(mockedZK).getData(anyString(), anyBoolean(),
    any(AsyncCallback.DataCallback.class), any());
  doAnswer(i -> null).when(mockedZK).close();
  when(mockedZK.getState()).thenReturn(ZooKeeper.States.CONNECTED);
  RO_ZK.zookeeper = mockedZK;
  CompletableFuture<byte[]> future = RO_ZK.get(PATH);
  AsyncCallback.DataCallback callback = exchanger.exchange(null);
  // 2 * keep alive time to ensure that we will not close the zk when there are pending requests
  Thread.sleep(6000);
  assertNotNull(RO_ZK.zookeeper);
  verify(mockedZK, never()).close();
  callback.processResult(Code.OK.intValue(), PATH, null, DATA, null);
  assertArrayEquals(DATA, future.get());
  // now we will close the idle connection.
  waitForIdleConnectionClosed();
  verify(mockedZK, times(1)).close();
}

@Override
protected void clientClosed(RMIConnection conn) throws IOException {
    System.out.println("clientClosed, will call connectorServer.stop");
    final Exchanger<Void> x = new Exchanger<Void>();
    Thread t = new Thread() {
        public void run() {
            try {
                connectorServer.stop();
            } catch (Exception e) {
                fail(e);
            }
        }
    };
    t.setName("connectorServer.stop");
    t.start();
    waitForBlock(t);
    /* If this thread is synchronized on RMIServerImpl, then
     * the thread that does connectorServer.stop will acquire
     * the clientList lock and then block waiting for the RMIServerImpl
     * lock.  Our call to super.clientClosed will then deadlock because
     * it needs to acquire the clientList lock.
     */
    System.out.println("calling super.clientClosed");
    System.out.flush();
    super.clientClosed(conn);
}

@Override
protected void clientClosed(RMIConnection conn) throws IOException {
    System.out.println("clientClosed, will call connectorServer.stop");
    final Exchanger<Void> x = new Exchanger<Void>();
    Thread t = new Thread() {
        public void run() {
            try {
                connectorServer.stop();
            } catch (Exception e) {
                fail(e);
            }
        }
    };
    t.setName("connectorServer.stop");
    t.start();
    waitForBlock(t);
    /* If this thread is synchronized on RMIServerImpl, then
     * the thread that does connectorServer.stop will acquire
     * the clientList lock and then block waiting for the RMIServerImpl
     * lock.  Our call to super.clientClosed will then deadlock because
     * it needs to acquire the clientList lock.
     */
    System.out.println("calling super.clientClosed");
    System.out.flush();
    super.clientClosed(conn);
}

@Override
protected void clientClosed(RMIConnection conn) throws IOException {
    System.out.println("clientClosed, will call connectorServer.stop");
    final Exchanger<Void> x = new Exchanger<Void>();
    Thread t = new Thread() {
        public void run() {
            try {
                connectorServer.stop();
            } catch (Exception e) {
                fail(e);
            }
        }
    };
    t.setName("connectorServer.stop");
    t.start();
    waitForBlock(t);
    /* If this thread is synchronized on RMIServerImpl, then
     * the thread that does connectorServer.stop will acquire
     * the clientList lock and then block waiting for the RMIServerImpl
     * lock.  Our call to super.clientClosed will then deadlock because
     * it needs to acquire the clientList lock.
     */
    System.out.println("calling super.clientClosed");
    System.out.flush();
    super.clientClosed(conn);
}

private synchronized boolean startCapture(
    final int width, final int height,
    final int min_mfps, final int max_mfps) {
  Log.d(TAG, "startCapture: " + width + "x" + height + "@" +
      min_mfps + ":" + max_mfps);
  if (cameraThread != null || cameraThreadHandler != null) {
    throw new RuntimeException("Camera thread already started!");
  }
  Exchanger<Handler> handlerExchanger = new Exchanger<Handler>();
  cameraThread = new CameraThread(handlerExchanger);
  cameraThread.start();
  cameraThreadHandler = exchange(handlerExchanger, null);

  final Exchanger<Boolean> result = new Exchanger<Boolean>();
  cameraThreadHandler.post(new Runnable() {
      @Override public void run() {
        startCaptureOnCameraThread(width, height, min_mfps, max_mfps, result);
      }
    });
  boolean startResult = exchange(result, false); // |false| is a dummy value.
  orientationListener.enable();
  return startResult;
}

private synchronized boolean stopCapture() {
  Log.d(TAG, "stopCapture");
  orientationListener.disable();
  final Exchanger<Boolean> result = new Exchanger<Boolean>();
  cameraThreadHandler.post(new Runnable() {
      @Override public void run() {
        stopCaptureOnCameraThread(result);
      }
    });
  boolean status = exchange(result, false);  // |false| is a dummy value here.
  try {
    cameraThread.join();
  } catch (InterruptedException e) {
    throw new RuntimeException(e);
  }
  cameraThreadHandler = null;
  cameraThread = null;
  Log.d(TAG, "stopCapture done");
  return status;
}

private void setPreviewRotationOnCameraThread(
    int rotation, Exchanger<IOException> result) {
  Log.v(TAG, "setPreviewRotation:" + rotation);

  int resultRotation = 0;
  if (info.facing == Camera.CameraInfo.CAMERA_FACING_FRONT) {
    // This is a front facing camera.  SetDisplayOrientation will flip
    // the image horizontally before doing the rotation.
    resultRotation = ( 360 - rotation ) % 360; // Compensate for the mirror.
  } else {
    // Back-facing camera.
    resultRotation = rotation;
  }
  camera.setDisplayOrientation(resultRotation);
  exchange(result, null);
}

@Override
public synchronized void surfaceCreated(final SurfaceHolder holder) {
  Log.d(TAG, "VideoCaptureAndroid::surfaceCreated");
  if (camera == null || cameraThreadHandler == null) {
    return;
  }
  final Exchanger<IOException> result = new Exchanger<IOException>();
  cameraThreadHandler.post(new Runnable() {
      @Override public void run() {
        setPreviewDisplayOnCameraThread(holder, result);
      }
    });
  IOException e = exchange(result, null);  // |null| is a dummy value here.
  if (e != null) {
    throw new RuntimeException(e);
  }
}

@Override
public synchronized void surfaceDestroyed(SurfaceHolder holder) {
  Log.d(TAG, "VideoCaptureAndroid::surfaceDestroyed");
  if (camera == null || cameraThreadHandler == null) {
    return;
  }
  final Exchanger<IOException> result = new Exchanger<IOException>();
  cameraThreadHandler.post(new Runnable() {
      @Override public void run() {
        setPreviewDisplayOnCameraThread(null, result);
      }
    });
  IOException e = exchange(result, null);  // |null| is a dummy value here.
  if (e != null) {
    throw new RuntimeException(e);
  }
}

@Override
protected void clientClosed(RMIConnection conn) throws IOException {
    System.out.println("clientClosed, will call connectorServer.stop");
    final Exchanger<Void> x = new Exchanger<Void>();
    Thread t = new Thread() {
        public void run() {
            try {
                connectorServer.stop();
            } catch (Exception e) {
                fail(e);
            }
        }
    };
    t.setName("connectorServer.stop");
    t.start();
    waitForBlock(t);
    /* If this thread is synchronized on RMIServerImpl, then
     * the thread that does connectorServer.stop will acquire
     * the clientList lock and then block waiting for the RMIServerImpl
     * lock.  Our call to super.clientClosed will then deadlock because
     * it needs to acquire the clientList lock.
     */
    System.out.println("calling super.clientClosed");
    System.out.flush();
    super.clientClosed(conn);
}

@Override
protected void doSendMessage(Message message) throws IOException {
    // the http stack is intended to send back only coap responses

    // check if the message is a response
    if (message instanceof Response) {
        // retrieve the request linked to the response
        Response response = (Response) message;
        Request request = response.getRequest();
        LOG.info("Handling response for request: " + request);

        // fill the exchanger with the incoming response
        Exchanger<Response> exchanger = exchangeMap.get(request);
        try {
            exchanger.exchange(response);
        } catch (InterruptedException e) {
            LOG.warning("Exchange interrupted: " + e.getMessage());

            // remove the entry from the map
            exchangeMap.remove(request);
            return;
        }

        LOG.info("Exchanged correctly");
    }
}

private JEditorPane paneFor(FileObject src, String fileName, String code) throws Exception, DataObjectNotFoundException, IOException {
    FileObject fromFO = FileUtil.createData(src, fileName);
    TestUtilities.copyStringToFile(fromFO, code);
    DataObject od = DataObject.find(fromFO);
    final EditorCookie.Observable ec = od.getCookie(EditorCookie.Observable.class);
    final Exchanger<JEditorPane> exch = new Exchanger<>();

    class L implements PropertyChangeListener {

        @Override
        public void propertyChange(PropertyChangeEvent evt) {
            try {
                if (!EditorCookie.Observable.PROP_OPENED_PANES.equals(evt.getPropertyName())) {
                    return;
                }
                // we are in AWT
                JEditorPane[] panes = ec.getOpenedPanes();
                if (panes == null) {
                    return;
                }
                exch.exchange(panes[0]);
            } catch (InterruptedException ex) {
            }
        }
    }
    L listener = new L();
    ec.addPropertyChangeListener(listener);
    JEditorPane pane = null;
    try {
        ec.open();
        ec.openDocument().putProperty(Language.class, JavaTokenId.language());
        pane = exch.exchange(null, 5, TimeUnit.SECONDS);
    } finally {
        ec.removePropertyChangeListener(listener);
    }
    assertNotNull("Editor pane not opened", pane);
    return pane;
}

public ClientPlatformConsumer(final ExecutorService executor,
        final Exchanger<GenericCommand<PlatformUnitCommandEnum>> exchanger,
        final Map<String, GenericMotor> engineCache) {
    this.executor = executor;
    this.exchanger = exchanger;
    this.rightMotor = engineCache.get(RIGHT);
    this.leftMotor = engineCache.get(LEFT);

}