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

@Test
public void testConsumerSingleMessage() throws Exception {
  TransferQueue<RabbitMessage> messages = new LinkedTransferQueue<>();

  Channel channel = mock(Channel.class);

  final Consumer consumer = new StreamSetsMessageConsumer(channel, messages);
  final Envelope envelope = new Envelope(1L, false, EXCHANGE_NAME, QUEUE_NAME);

  executor.submit(new Runnable() {
    @Override
    public void run() {
      try {
        consumer.handleDelivery("consumerTag", envelope, null, TEST_MESSAGE_1.getBytes());
      } catch (IOException ignored) {
        // no op
      }
    }
  });

  RabbitMessage message = messages.take();
  assertEquals(TEST_MESSAGE_1, new String(message.getBody(), StandardCharsets.UTF_8));
}

public BasicBlockingQueueSender(
        final String name,
        final int batchSize,
        final BlockingQueue<Object> queue,
        final boolean checkBusy,
        final UnableToEnqueueHandler unableToEnqueueHandler,
        final Queue<T> owner) {

    super(queue, owner, batchSize, name + "| BQ SEND QUEUE", LoggerFactory.getLogger(BasicBlockingQueueSender.class));


    this.unableToEnqueueHandler = unableToEnqueueHandler;
    if (queue instanceof TransferQueue && checkBusy) {
        throw new IllegalStateException("Should never pass transfer queue");
    }
}

Runnable newLinkedTransferQueueRunner() {
  final TransferQueue<Integer> queue = new LinkedTransferQueue<>();
  return new Runnable() {
    @Override public void run() {
      final ThreadLocalRandom random = ThreadLocalRandom.current();
      for (;;) {
        if (random.nextBoolean()) {
          queue.offer(ELEMENT);
        } else {
          queue.poll();
        }
        calls.increment();
      }
    }
  };
}

private int transfer() {
    try {
        ((TransferQueue)q).tryTransfer(1, 1, TimeUnit.SECONDS);
    } catch (InterruptedException ie) {
    }
    return 1;
}

public BasicSendQueueWithTryTransfer(
        final String name,
        final int batchSize,
        final TransferQueue<Object> queue,
        final int checkBusyEvery,
        final Queue<T> owner) {

    super(queue, owner, batchSize, name + "| TQT SEND QUEUE", LoggerFactory.getLogger(BasicSendQueueWithTryTransfer.class));


    this.queue = queue;
    this.checkBusyEvery = checkBusyEvery;
}

public BasicSendQueueWithTransferQueue(
        final String name,
        final int batchSize,
        final TransferQueue<Object> queue,
        final int checkBusyEvery,
        final Queue<T> owner) {

    super(queue, owner, batchSize, name + "| TQ SEND QUEUE", LoggerFactory.getLogger(BasicSendQueueWithTransferQueue.class));


    this.queue = queue;
    this.checkBusyEvery = checkBusyEvery;
}

TransferQueue<RabbitMessage> getMessageQueue() {
  return messages;
}

@Test
public void testHeaderProcessing() throws Exception {
  ((RabbitSourceConfigBean)conf).basicConfig.maxWaitTime = 1000; // Set this low so that we don't slow down the test.

  stage = PowerMockito.spy(newStage());

  // setup some fake data and force it onto the source's queue
  RabbitSource source = (RabbitSource)stage;
  TransferQueue<RabbitMessage> messages = source.getMessageQueue();
  Envelope envelope = new Envelope(DELIVERY_TAG, REDELIVERED, EXCHANGE_NAME, QUEUE_NAME);
  AMQP.BasicProperties.Builder propertiesBuilder = new AMQP.BasicProperties.Builder();
  propertiesBuilder.contentType(CONTENT_TYPE);
  Map<String, Object> customHeaders = new HashMap<>();
  customHeaders.put(CUSTOM_HEADER_KEY, CUSTOM_HEADER_VAL);
  propertiesBuilder.headers(customHeaders);
  propertiesBuilder.clusterId(CLUSTER_ID);
  AMQP.BasicProperties properties = propertiesBuilder.build();
  RabbitMessage msg = new RabbitMessage(CONSUMER_TAG, envelope, properties, TEST_MESSAGE_1.getBytes());
  source.getMessageQueue().put(msg);
  doReturn(new ArrayList<Stage.ConfigIssue>()).when((RabbitSource)stage).init();

  PowerMockito.doReturn(false).when(stage, "isConnected");

  this.runner = newStageRunner("output");

  // setup items which are not correctly configured in init
  Channel channel = mock(Channel.class);
  StreamSetsMessageConsumer consumer = new StreamSetsMessageConsumer(channel, messages);
  source.setStreamSetsMessageConsumer(consumer);
  DataParserFactory parserFactory = new DataParserFactoryBuilder(runner.getContext(), DataParserFormat.JSON)
      .setCharset(StandardCharsets.UTF_8)
      .setMode(JsonMode.MULTIPLE_OBJECTS)
      .setMaxDataLen(-1)
      .build();
  source.setDataParserFactory(parserFactory);

  runner.runInit();

  StageRunner.Output output = ((SourceRunner)runner).runProduce(null, 1000);
  List<Record> records = output.getRecords().get("output");
  assertEquals(1, records.size());
  Record record = records.get(0);
  assertEquals(String.valueOf(DELIVERY_TAG), record.getHeader().getAttribute("deliveryTag"));
  assertEquals(String.valueOf(REDELIVERED), record.getHeader().getAttribute("redelivered"));
  assertEquals(EXCHANGE_NAME, record.getHeader().getAttribute("exchange"));
  assertEquals(CONTENT_TYPE, record.getHeader().getAttribute("contentType"));
  assertNull(record.getHeader().getAttribute("appId"));
  assertEquals(CUSTOM_HEADER_VAL, record.getHeader().getAttribute(CUSTOM_HEADER_KEY));
  runner.runDestroy();
}

public TransferQueue<String> getTransferQueue() {
    return this.transferQueue;
}

public void setTransferQueue(TransferQueue<String> transferQueue) {
    this.transferQueue = transferQueue;
}

public static void main(String[] args) {
    TransferQueue<String> tq = new LinkedTransferQueue<>();
    BlockingQueue<String> bq =new  LinkedBlockingQueue<>(100);
}

/**
 * @param manager
 * @throws IOException 
 */
public TCPSender(NetworkManager manager,SocketChannel socket,TransferQueue<Message> queue) throws IOException{
    this.manager = manager;
    this.client = socket;
    this.send_queue = queue;
}

public TransferQueue<Promise> getPromiseQueue(){
    return promises;
}

/**
 * @param values
 */
public Consumer(TransferQueue<Long> values) {
    this.values = values;
}

/**
 * @param values
 */
public Producer(TransferQueue<Long> values) {
    this.values = values;
}

private boolean transfer(final LogEvent memento) {
    return queue instanceof TransferQueue
        ? ((TransferQueue<LogEvent>) queue).tryTransfer(memento)
        : queue.offer(memento);
}

/**
 * Constructs a new instance and records its association to the passed-in channel.
 *
 * @param channel the channel to which this consumer is attached
 */
public StreamSetsMessageConsumer(Channel channel, TransferQueue<RabbitMessage> records) {
  super(channel);
  this.records = records;
}