java.util.Queue#forEach ( )源码实例Demo

/**
 * Verifies bucket table data is matching with actual response added to
 * double buffer.
 */
private void checkCreateBuckets(Queue<OMBucketCreateResponse> bucketQueue) {
  bucketQueue.forEach((omBucketCreateResponse) -> {
    OmBucketInfo omBucketInfo = omBucketCreateResponse.getOmBucketInfo();
    String bucket = omBucketInfo.getBucketName();
    OmBucketInfo tableBucketInfo = null;
    try {
      tableBucketInfo =
          omMetadataManager.getBucketTable().get(
              omMetadataManager.getBucketKey(omBucketInfo.getVolumeName(),
                  bucket));
    } catch (IOException ex) {
      fail("testDoubleBufferWithMixOfTransactions failed");
    }
    Assert.assertNotNull(tableBucketInfo);

    Assert.assertEquals(omBucketInfo.getVolumeName(),
        tableBucketInfo.getVolumeName());
    Assert.assertEquals(omBucketInfo.getBucketName(),
        tableBucketInfo.getBucketName());
    Assert.assertEquals(omBucketInfo.getCreationTime(),
        tableBucketInfo.getCreationTime());
  });
}
 

public synchronized void clearTerminatedJobs() {
    Iterator<Map.Entry<Long, UpstreamJob>> itr = workingJobs.entrySet().iterator();
    while (itr.hasNext()) {
        UpstreamJob job = itr.next().getValue();
        if (job.isTerminated()) {
            job.terminate();
            itr.remove();
        }
    }
    Queue<UpstreamJob> oldQueue = queue;
    queue = new ArrayDeque<>();
    oldQueue.forEach(job -> {
        if (!job.isTerminated()) {
            queue.add(job);
        }
    });
}
 

public synchronized void clearTerminatedJobs() {
    Iterator<Map.Entry<Long, UpstreamJob>> itr = workingJobs.entrySet().iterator();
    while (itr.hasNext()) {
        UpstreamJob job = itr.next().getValue();
        if (job.isTerminated()) {
            job.terminate();
            itr.remove();
        }
    }
    Queue<UpstreamJob> oldQueue = queue;
    queue = new ArrayDeque<>();
    oldQueue.forEach(job -> {
        if (!job.isTerminated()) {
            queue.add(job);
        }
    });
}
 

/**
 * Adds an initial step to this walking queue.
 *
 * @param step The step to add.
 */
public void addFirst(Step step) {
    current.clear();
    runningPath = false;

    Queue<Step> backtrack = new ArrayDeque<>();
    for (; ; ) {
        Step prev = previous.pollLast();
        if (prev == null) {
            break;
        }
        backtrack.add(prev);
        if (prev.equals(step)) {
            backtrack.forEach(this::add);
            previous.clear();
            return;
        }
    }
    previous.clear();

    add(step);
}
 

/**
 * Verifies deleted bucket responses added to double buffer are actually
 * removed from the OM DB or not.
 */
private void checkDeletedBuckets(Queue<OMBucketDeleteResponse>
    deleteBucketQueue) {
  deleteBucketQueue.forEach((omBucketDeleteResponse -> {
    try {
      Assert.assertNull(omMetadataManager.getBucketTable().get(
          omMetadataManager.getBucketKey(
              omBucketDeleteResponse.getVolumeName(),
              omBucketDeleteResponse.getBucketName())));
    } catch (IOException ex) {
      fail("testDoubleBufferWithMixOfTransactions failed");
    }
  }));
}
 

private void register(ServerAddress remoteAddress, UUID uuid, ActorRef sender) {
  sessionActors.put(remoteAddress, new SessionActorInfo(uuid, sender));
  log.debug("[{}][{}] Registering session actor.", remoteAddress, uuid);
  Queue<ClusterAPIProtos.ToRpcServerMessage> data = pendingMsgs.remove(remoteAddress);
  if (data != null) {
    log.debug("[{}][{}] Forwarding {} pending messages.", remoteAddress, uuid, data.size());
    data.forEach(msg -> sender.tell(new RpcSessionTellMsg(remoteAddress, msg), ActorRef.noSender()));
  } else {
    log.debug("[{}][{}] No pending messages to forward.", remoteAddress, uuid);
  }
}
 

/**
     * @inheritDoc
     */
    @Override
    public void reroot(Ligand<A> ligand) {

//        System.out.println("tails: " + arcs.tails);
//        System.out.println("heads: " + arcs.heads);
        root = ligand;
        ligand.reset();

        Queue<Arc<A>> queue = new LinkedList<>();

        // get parent arcs
        Arc<A> arc = arcs.getForHead(ligand);
        while (arc != null) {
            arcs.remove(arc);
            Arc<A> next = arcs.getForHead(arc.getTail());
            arc.transpose();
            queue.add(arc);
            arc = next;
        }

        queue.forEach((transposedArc) -> {
            arcs.add(transposedArc);
        });

        ligand.setParent(ligand.getAtom());

    }
 

public List<UUID> getActorsFromAlias(String alias) {
	List<UUID> result = new LinkedList<>();
	
	Queue<UUID> queue = aliases.get(alias);
	if (queue!=null)
		queue.forEach((id) -> result.add(id));
	
	return result;
}
 

@Test
void defaultStartStopExecutor() {
    final Server server = ServerTest.server.server();
    final Queue<Thread> threads = new LinkedTransferQueue<>();
    server.addListener(new ThreadRecordingServerListener(threads));

    threads.add(server.stop().thenApply(unused -> Thread.currentThread()).join());
    threads.add(server.start().thenApply(unused -> Thread.currentThread()).join());

    threads.forEach(t -> assertThat(t.getName()).startsWith("globalEventExecutor"));
}
 

@Test
void customStartStopExecutor() {
    final Queue<Thread> threads = new LinkedTransferQueue<>();
    final String prefix = getClass().getName() + "#customStartStopExecutor";

    final AtomicBoolean serverStarted = new AtomicBoolean();
    final ThreadFactory factory = ThreadFactories.builder(prefix).taskFunction(task -> () -> {
        await().untilFalse(serverStarted);
        task.run();
    }).build();

    final ExecutorService executor = Executors.newSingleThreadExecutor(factory);
    final Server server = Server.builder()
                                .startStopExecutor(executor)
                                .service("/", (ctx, req) -> HttpResponse.of(200))
                                .serverListener(new ThreadRecordingServerListener(threads))
                                .build();

    threads.add(server.start().thenApply(unused -> Thread.currentThread()).join());
    serverStarted.set(true);

    final CompletableFuture<Thread> stopFuture = server.stop().thenApply(
            unused -> Thread.currentThread());
    serverStarted.set(false);
    threads.add(stopFuture.join());

    threads.forEach(t -> assertThat(t.getName()).startsWith(prefix));
}
 

public void test_Queue_forEach() {
    Deque<Integer> deque = new ArrayDeque<Integer>();
    deque.addFirst(2);
    deque.addFirst(1);
    deque.addFirst(0);

    Queue<Integer> queue = Collections.asLifoQueue(deque);
    ArrayList<Integer> output = new ArrayList<Integer>();
    queue.forEach(v -> output.add(v));

    assertEquals(3, output.size());
    assertEquals(0, (int)output.get(0));
    assertEquals(1, (int)output.get(1));
    assertEquals(2, (int)output.get(2));
}
 

/**
 * Activates the given bucket number.
 *
 * @param bucket the bucket number to activate
 */
private void activateBucket(int bucket) {
    Queue<Runnable> tasks = flowTasks.remove(bucket);
    if (tasks != null) {
        log.debug("Completing enqueued operations for device {}", deviceId);
        tasks.forEach(task -> task.run());
    }
}
 

private static void runLocal(final String[] args, final ClassLoader systemClassLoader, final File narWorkingDirectory) throws Exception {
    final boolean once = args[1].equalsIgnoreCase("Once");

    final String json;
    if (args[2].equals(FILE_FLAG)) {
        json = new String(Files.readAllBytes(Paths.get(args[3])));
    } else if (args[2].equals(JSON_FLAG)) {
        json = args[3];
    } else if (args[2].equals(YARN_JSON_FLAG)) {
        json = args[3].replace(';', ',');
    } else {
        System.out.println("Invalid input: " + formatArgs(args));
        printUsage();
        System.exit(1);
        return;
    }

    JsonObject jsonObject = new JsonParser().parse(json).getAsJsonObject();
    System.out.println("Running from json: " + StatelessSecurityUtility.formatJson(jsonObject));
    final RunnableFlow flow = StatelessFlow.createAndEnqueueFromJSON(jsonObject, systemClassLoader, narWorkingDirectory);

    // Run Flow
    final Queue<InMemoryFlowFile> outputFlowFiles = new LinkedList<>();
    final boolean successful;
    if (once) {
        successful = flow.runOnce(outputFlowFiles);
    } else {
        successful = flow.run(outputFlowFiles);  //Run forever
    }

    // TODO: Introduce verbose flag to determine if flowfiles should be printed on completion
    if (successful) {
        System.out.println("Flow Succeeded");
        if (isVerbose) {
            outputFlowFiles.forEach(f -> System.out.println(f.toStringFull()));
        }
    } else {
        System.out.println("Flow Failed");
        if (isVerbose) {
            outputFlowFiles.forEach(f -> System.out.println(f.toStringFull()));
        }
        System.exit(1);
    }
}
 

@Test
public void givenQueue_thenIterateAndPrintResults() {
    Queue<String> namesQueue = new ArrayDeque<>(Arrays.asList("Larry", "Steve", "James"));

    namesQueue.forEach(System.out::println);
}