java类com.google.common.collect.Queues的实例源码

@Test(dependsOnMethods = "testSerializeToSequenceFile")
public void testDeserializeFromSequenceFile() throws IOException {
  Queue<WorkUnitState> workUnitStates = Queues.newConcurrentLinkedQueue();

  Closer closer = Closer.create();
  try {
    ParallelRunner parallelRunner = closer.register(new ParallelRunner(2, this.fs));
    parallelRunner.deserializeFromSequenceFile(Text.class, WorkUnitState.class, new Path(this.outputPath, "seq1"),
        workUnitStates);
    parallelRunner.deserializeFromSequenceFile(Text.class, WorkUnitState.class, new Path(this.outputPath, "seq2"),
        workUnitStates);
  } catch (Throwable t) {
    throw closer.rethrow(t);
  } finally {
    closer.close();
  }

  Assert.assertEquals(workUnitStates.size(), 2);

  for (WorkUnitState workUnitState : workUnitStates) {
    TestWatermark watermark = new Gson().fromJson(workUnitState.getActualHighWatermark(), TestWatermark.class);
    Assert.assertTrue(watermark.getLongWatermark() == 10L || watermark.getLongWatermark() == 100L);
  }
}

public EventReporter(Builder builder) {
  super(builder.context, builder.name, builder.filter, builder.rateUnit, builder.durationUnit);

  this.closer = Closer.create();
  this.immediateReportExecutor = MoreExecutors.
      getExitingExecutorService((ThreadPoolExecutor) Executors.newFixedThreadPool(1,
          ExecutorsUtils.newThreadFactory(Optional.of(LOGGER), Optional.of("EventReporter-" + builder.name + "-%d"))),
          5, TimeUnit.MINUTES);

  this.metricContext = builder.context;
  this.notificationTargetKey = builder.context.addNotificationTarget(new Function<Notification, Void>() {
    @Nullable
    @Override
    public Void apply(Notification notification) {
      notificationCallback(notification);
      return null;
    }
  });
  this.reportingQueue = Queues.newLinkedBlockingQueue(QUEUE_CAPACITY);
}

protected void clear(Commit<Void> commit) {
  Queue<DocumentPath> toClearQueue = Queues.newArrayDeque();
  Map<String, Versioned<byte[]>> topLevelChildren = docTree.getChildren(DocumentPath.from("root"));
  toClearQueue.addAll(topLevelChildren.keySet()
      .stream()
      .map(name -> new DocumentPath(name, DocumentPath.from("root")))
      .collect(Collectors.toList()));
  while (!toClearQueue.isEmpty()) {
    DocumentPath path = toClearQueue.remove();
    Map<String, Versioned<byte[]>> children = docTree.getChildren(path);
    if (children.size() == 0) {
      docTree.removeNode(path);
    } else {
      children.keySet().forEach(name -> toClearQueue.add(new DocumentPath(name, path)));
      toClearQueue.add(path);
    }
  }
}

@Nonnull
Iterator<NodeDocument> getAllPreviousDocs() {
    if (getPreviousRanges().isEmpty()) {
        return Iterators.emptyIterator();
    }
    //Currently this method would fire one query per previous doc
    //If that poses a problem we can try to find all prev doc by relying
    //on property that all prevDoc id would starts <depth+2>:p/path/to/node
    return new AbstractIterator<NodeDocument>(){
        private Queue<Map.Entry<Revision, Range>> previousRanges =
                Queues.newArrayDeque(getPreviousRanges().entrySet());
        @Override
        protected NodeDocument computeNext() {
            if(!previousRanges.isEmpty()){
                Map.Entry<Revision, Range> e = previousRanges.remove();
                NodeDocument prev = getPreviousDoc(e.getKey(), e.getValue());
                if(prev != null){
                    previousRanges.addAll(prev.getPreviousRanges().entrySet());
                    return prev;
                }
            }
            return endOfData();
        }
    };
}

@Test
public void testForwardingOfRequests() throws Exception {

  Queue<RequestAndCallback> queue = Queues.newArrayDeque();

  BatchedPermitsRequester container = BatchedPermitsRequester.builder().resourceId("resource")
      .requestorIdentifier("requestor").requestSender(new TestRequestSender(queue, false)).build();
  try (ParallelRequester requester = new ParallelRequester(container)) {

    Future<Boolean> future = requester.request(10);

    await(new QueueSize(queue, 1), 1000);
    Assert.assertEquals(queue.size(), 1);
    satisfyRequestBuilder().requestAndCallback(queue.poll()).satisfy();

    future.get(1, TimeUnit.SECONDS);
    Assert.assertTrue(future.isDone());
    Assert.assertTrue(future.get());
  }
}

@Test
public void testRetriableFail() throws Exception {
  Queue<RequestAndCallback> queue = Queues.newArrayDeque();

  BatchedPermitsRequester container = BatchedPermitsRequester.builder().resourceId("resource")
      .requestorIdentifier("requestor").requestSender(new TestRequestSender(queue, false)).build();
  try (ParallelRequester requester = new ParallelRequester(container)) {

    Future<Boolean> future = requester.request(10);

    for (int i = 0; i < BatchedPermitsRequester.MAX_RETRIES; i++) {
      // container will fail 5 times
      await(new QueueSize(queue, 1), 1000);
      Assert.assertFalse(future.isDone());
      failRequestBuilder().requestAndCallback(queue.poll()).fail();
    }

    // should return a failure
    Assert.assertFalse(future.get());
    // should not make any more request
    Assert.assertEquals(queue.size(), 0);
  }
}

@Test
public void testNonRetriableFail() throws Exception {
  Queue<RequestAndCallback> queue = Queues.newArrayDeque();

  BatchedPermitsRequester container = BatchedPermitsRequester.builder().resourceId("resource")
      .requestorIdentifier("requestor").requestSender(new TestRequestSender(queue, false)).build();
  try (ParallelRequester requester = new ParallelRequester(container)) {

    Future<Boolean> future = requester.request(10);

    // container should only try request once
    await(new QueueSize(queue, 1), 1000);
    Assert.assertFalse(future.isDone());
    failRequestBuilder().requestAndCallback(queue.poll()).errorStatus(HttpStatus.S_422_UNPROCESSABLE_ENTITY).fail();

    Assert.assertFalse(future.get());
    Assert.assertEquals(queue.size(), 0);
  }
}

@Test(dependsOnMethods = "testSerializeToSequenceFile")
public void testDeserializeFromSequenceFile() throws IOException {
  Queue<WorkUnitState> workUnitStates = Queues.newConcurrentLinkedQueue();

  Path seqPath1 = new Path(this.outputPath, "seq1");
  Path seqPath2 = new Path(this.outputPath, "seq2");

  try (ParallelRunner parallelRunner = new ParallelRunner(2, this.fs)) {
    parallelRunner.deserializeFromSequenceFile(Text.class, WorkUnitState.class, seqPath1, workUnitStates, true);
    parallelRunner.deserializeFromSequenceFile(Text.class, WorkUnitState.class, seqPath2, workUnitStates, true);
  }

  Assert.assertFalse(this.fs.exists(seqPath1));
  Assert.assertFalse(this.fs.exists(seqPath2));

  Assert.assertEquals(workUnitStates.size(), 2);

  for (WorkUnitState workUnitState : workUnitStates) {
    TestWatermark watermark = new Gson().fromJson(workUnitState.getActualHighWatermark(), TestWatermark.class);
    Assert.assertTrue(watermark.getLongWatermark() == 10L || watermark.getLongWatermark() == 100L);
  }
}

public EventReporter(Builder builder) {
  super(builder.context, builder.name, builder.filter, builder.rateUnit, builder.durationUnit);

  this.closer = Closer.create();
  this.immediateReportExecutor = MoreExecutors.getExitingExecutorService(
      (ThreadPoolExecutor) Executors.newFixedThreadPool(1,
          ExecutorsUtils.newThreadFactory(Optional.of(LOGGER), Optional.of("EventReporter-" + builder.name + "-%d"))),
      5, TimeUnit.MINUTES);

  this.metricContext = builder.context;
  this.notificationTargetKey = builder.context.addNotificationTarget(new Function<Notification, Void>() {
    @Nullable
    @Override
    public Void apply(Notification notification) {
      notificationCallback(notification);
      return null;
    }
  });
  this.reportingQueue = Queues.newLinkedBlockingQueue(QUEUE_CAPACITY);
}

SelectDispatcher(SelectConfig config) {
    super(config);

    m_config = config;

    CassandraSession session = new CassandraSessionImpl(
            config.getCassandraKeyspace(),
            config.getCassandraHost(),
            config.getCassandraPort(),
            config.getCassandraCompression(),
            config.getCassandraUsername(),
            config.getCassandraPassword(),
            config.getCassandraSsl());
    m_repository = new CassandraSampleRepository(
            session,
            Config.CASSANDRA_TTL,
            new MetricRegistry(),
            new DefaultSampleProcessorService(1),
            new ContextConfigurations());

    m_queryQueue = Queues.newArrayBlockingQueue(config.getThreads() * 10);

}