java.util.concurrent.atomic.AtomicReference#set()源码实例Demo

@Test
public void testYield() throws Exception {
	final AtomicReference<Exception> exceptionReference = new AtomicReference<>();
	final TestRunnable testRunnable = new TestRunnable();
	final Thread submitThread = new Thread(() -> {
		try {
			mailboxExecutorService.execute(testRunnable);
		} catch (Exception e) {
			exceptionReference.set(e);
		}
	});

	submitThread.start();
	mailboxExecutorService.yield();
	submitThread.join();

	Assert.assertNull(exceptionReference.get());
	Assert.assertEquals(Thread.currentThread(), testRunnable.wasExecutedBy());
}
 

private static Subscriber<String> OBSERVER_ONNEXT_FAIL(final AtomicReference<Throwable> onError) {
    return new DefaultSubscriber<String>() {

        @Override
        public void onComplete() {

        }

        @Override
        public void onError(Throwable e) {
            onError.set(e);
        }

        @Override
        public void onNext(String args) {
            throw new SafeSubscriberTestException("onNextFail");
        }
    };

}
 

/**
 * Actually create a new WebDriver session. The returned webdriver is not guaranteed to be a
 * {@link RemoteWebDriver}.
 */
public WebDriver build() {
  if (options.isEmpty() && additionalCapabilities.isEmpty()) {
    throw new SessionNotCreatedException("Refusing to create session without any capabilities");
  }

  Plan plan = getPlan();

  CommandExecutor executor;
  if (plan.isUsingDriverService()) {
    AtomicReference<DriverService> serviceRef = new AtomicReference<>();

    executor = new SpecCompliantExecutor(
        () -> {
          if (serviceRef.get() != null && serviceRef.get().isRunning()) {
            throw new SessionNotCreatedException(
                "Attempt to start the underlying service more than once");
          }
          try {
            DriverService service = plan.getDriverService();
            serviceRef.set(service);
            service.start();
            return service.getUrl();
          } catch (IOException e) {
            throw new SessionNotCreatedException(e.getMessage(), e);
          }
        },
        plan::writePayload,
        () -> serviceRef.get().stop());
  } else {
    executor = new SpecCompliantExecutor(() -> remoteHost, plan::writePayload, () -> {});
  }

  return new RemoteWebDriver(executor, new ImmutableCapabilities());
}
 

@Test
public void testCancelLeftOuterJoinTaskWhileProbing() throws Exception {
	setOutput(new DiscardingOutputCollector<Tuple2<Integer, Integer>>());
	addDriverComparator(this.comparator1);
	addDriverComparator(this.comparator2);
	getTaskConfig().setDriverPairComparator(new RuntimePairComparatorFactory());
	getTaskConfig().setDriverStrategy(DriverStrategy.LEFT_HYBRIDHASH_BUILD_SECOND);
	getTaskConfig().setRelativeMemoryDriver(this.hash_frac);

	final AbstractOuterJoinDriver<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> testTask = getOuterJoinDriver();

	addInput(new DelayingIterator<>(new InfiniteIntTupleIterator(), 100), this.serializer);
	addInput(new UniformIntTupleGenerator(1, 1, true), this.serializer);

	final AtomicReference<Throwable> error = new AtomicReference<>();

	final Thread taskRunner = new Thread("Task runner for testCancelOuterJoinTaskWhileSort1()") {
		@Override
		public void run() {
			try {
				testDriver(testTask, MockJoinStub.class);
			} catch (Throwable t) {
				error.set(t);
			}
		}
	};
	taskRunner.start();

	Thread.sleep(1000);

	cancel();
	taskRunner.join(60000);

	assertFalse("Task thread did not finish within 60 seconds", taskRunner.isAlive());

	final Throwable taskError = error.get();
	if (taskError != null) {
		fail("Error in task while canceling:\n" + Throwables.getStackTraceAsString(taskError));
	}
}
 

@Test
public void pingPongParallelCall() throws Exception {
	Scheduler s = Schedulers.newParallel("work", 4);
	try {
		Scheduler.Worker w = s.createWorker();
		Thread t = Thread.currentThread();
		AtomicReference<Thread> t1 = new AtomicReference<>(t);
		AtomicReference<Thread> t2 = new AtomicReference<>(t);

		CountDownLatch latch = new CountDownLatch(4);

		AtomicReference<Runnable> pong = new AtomicReference<>();

		Runnable ping = () -> {
			if(latch.getCount() > 0){
				t1.set(Thread.currentThread());
				w.schedule(pong.get());
				latch.countDown();
			}
		};

		pong.set(() -> {
			if(latch.getCount() > 0){
				t2.set(Thread.currentThread());
				w.schedule(ping);
				latch.countDown();
			}
		});

		w.schedule(ping);

		latch.await();

		assertThat(t).isNotEqualTo(t1.get());
		assertThat(t).isNotEqualTo(t2.get());
	}
	finally {
		s.dispose();
	}
}
 

@Test(timeout = 1000)
public void batcher_blocks_when_queue_is_full() throws Exception {
  Function<List<Integer>, List<Integer>> batchHandler = Function.identity();
  AtomicReference<Runnable> processQueue = new AtomicReference<>();
  ThreadFactory factory = r -> {
    processQueue.set(r);
    return new Thread();
  };

  CountDownLatch latch = new CountDownLatch(1);

  Thread thread = null;
  try (MessageBatcher<Integer, Integer> batcher = new MessageBatcher<>(4, 2, EnqueueBehaviour.BLOCK_AND_WAIT,
      batchHandler, factory)) {
    thread = new Thread(() -> {
      batcher.apply(1);
      batcher.apply(2);
      batcher.apply(3);
      batcher.apply(4);
      latch.countDown();
      batcher.apply(5);
    });
    thread.start();
    latch.await();

    while (thread.getState() != State.WAITING) {}
  } finally {
    if (thread != null) {
      thread.interrupt();
      thread.join();
    }
  }
}
 

@Before
public void setUp() throws Exception {
    Field f = Lookout.class.getDeclaredField("atomicRegistryReference");
    f.setAccessible(true);
    AtomicReference atomicReference = (AtomicReference) f.get(null);
    atomicReference.set(NoopRegistry.INSTANCE);

}
 

@Test
public void testSignalUnAvailable() throws Exception {

	final AtomicInteger counter = new AtomicInteger(0);
	final AtomicReference<MailboxDefaultAction.Suspension> suspendedActionRef = new AtomicReference<>();
	final OneShotLatch actionSuspendedLatch = new OneShotLatch();
	final int blockAfterInvocations = 3;
	final int totalInvocations = blockAfterInvocations * 2;

	MailboxThread mailboxThread = new MailboxThread() {
		@Override
		public void runDefaultAction(Controller controller) {
			if (counter.incrementAndGet() == blockAfterInvocations) {
				suspendedActionRef.set(controller.suspendDefaultAction());
				actionSuspendedLatch.trigger();
			} else if (counter.get() == totalInvocations) {
				controller.allActionsCompleted();
			}
		}
	};

	MailboxProcessor mailboxProcessor = start(mailboxThread);
	actionSuspendedLatch.await();
	Assert.assertEquals(blockAfterInvocations, counter.get());

	MailboxDefaultAction.Suspension suspension = suspendedActionRef.get();
	mailboxProcessor.getMailboxExecutor(DEFAULT_PRIORITY).execute(suspension::resume, "resume");
	mailboxThread.join();
	Assert.assertEquals(totalInvocations, counter.get());
}
 

/**
 * Captures the request attributes. Useful for testing ClientCalls.
 * {@link ClientCall#getAttributes()}
 */
private static ClientInterceptor recordClientCallInterceptor(
    final AtomicReference<ClientCall<?, ?>> clientCallCapture) {
  return new ClientInterceptor() {
    @Override
    public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
        MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
      ClientCall<ReqT, RespT> clientCall = next.newCall(method,callOptions);
      clientCallCapture.set(clientCall);
      return clientCall;
    }
  };
}
 

/**
 * toString returns current value.
 */
public void testToString() {
    AtomicReference<Integer> ai = new AtomicReference<>(one);
    assertEquals(one.toString(), ai.toString());
    ai.set(two);
    assertEquals(two.toString(), ai.toString());
}
 

private static WatchImplBase createWatchImpBase(AtomicReference<StreamObserver<WatchResponse>> responseObserverRef,
    StreamObserver<WatchRequest> requestStreamObserver) {
    return new WatchImplBase() {
        @Override
        public StreamObserver<WatchRequest> watch(StreamObserver<WatchResponse> responseObserver) {
            responseObserverRef.set(responseObserver);
            return requestStreamObserver;
        }
    };
}
 

public static void main(String[] args) throws Exception {
    ThreadGroup stubTG = new ThreadGroup(getRootThreadGroup(), "Stub Thread Group");
    ThreadGroup swingTG = new ThreadGroup(getRootThreadGroup(), "SwingTG");
    try {
        Thread stubThread = new Thread(stubTG, SunToolkit::createNewAppContext);
        stubThread.start();
        stubThread.join();

        CountDownLatch startSwingLatch = new CountDownLatch(1);
        new Thread(swingTG, () -> {
            SunToolkit.createNewAppContext();
            SwingUtilities.invokeLater(() -> {
                frame = new JFrame();
                component = new JLabel("Test Text");
                frame.add(component);
                frame.setBounds(100, 100, 100, 100);
                frame.setVisible(true);
                startSwingLatch.countDown();
            });
        }).start();
        startSwingLatch.await();

        AtomicReference<Exception> caughtException = new AtomicReference<>();
        Thread checkThread = new Thread(getRootThreadGroup(), () -> {
            try {
                component.invalidate();
                component.revalidate();
                component.repaint(new Rectangle(0, 0, 0, 0));
            } catch (Exception e) {
                caughtException.set(e);
            }
        });
        checkThread.start();
        checkThread.join();

        if (caughtException.get() != null) {
            throw new RuntimeException("Failed. Caught exception!", caughtException.get());
        }
    } finally {
        new Thread(swingTG, () -> SwingUtilities.invokeLater(() -> {
            if (frame != null) {
                frame.dispose();
            }
        })).start();
    }
}
 

@org.junit.Test
@ForAllEnvironments(section = "remote.platforms")
public void testCopyToRemote() throws Exception {
    ExecutionEnvironment execEnv = getTestExecutionEnvironment();
    assertNotNull(execEnv);
    File src = createTempFile("test-upload-1", null, false); // NOI18N
    src.deleteOnExit();
    writeFile(src, "qwe/nasd/nzxc"); // NOI18N
    String dst = "/tmp/" + /* execEnv.getUser() + "/" +  */ src.getName(); // NOI18N
    System.err.printf("testUploadFile: %s to %s:%s\n", src.getAbsolutePath(), execEnv.getDisplayName(), dst); // NOI18N

    Future<UploadStatus> uploadTask;
    int rc;

    uploadTask = CommonTasksSupport.uploadFile(src.getAbsolutePath(), execEnv, dst, 0755);
    UploadStatus uploadStatus = uploadTask.get();
    assertEquals("Error uploading " + src.getAbsolutePath() + " to " + execEnv + ":" + dst + ' ' + uploadStatus.getError(), 0, uploadStatus.getExitCode());
    assertTrue(HostInfoUtils.fileExists(execEnv, dst));

    StatInfo statFomrUpload = uploadStatus.getStatInfo();
    StatInfo stat = FileInfoProvider.lstat(execEnv, dst).get();
    assertEquals("Stat got from upload differ", stat.toExternalForm(), statFomrUpload.toExternalForm());

    final AtomicReference<Object> ref = new AtomicReference<>();
    CommonTasksSupport.UploadParameters up = new CommonTasksSupport.UploadParameters(
            src, execEnv, dst, null, 0755, false, new ChangeListener() {
        @Override
        public void stateChanged(ChangeEvent e) {
            ref.set(e.getSource());
        }
    });
    uploadTask = CommonTasksSupport.uploadFile(up);

    rc = uploadTask.get().getExitCode();

    // sleep a bit since listener can be just not calleds
    if (ref.get() == null) {
        sleep(100);
    }
    if (ref.get() == null) {
        sleep(500);
    }

    assertEquals("Error uploading " + src.getAbsolutePath() + " to " + execEnv + ":" + dst, 0, rc);

    assertNotNull("callback wasn't called", ref.get());
    assertEquals("callback was called with incorrect source object", uploadTask, ref.get());

    Future<Integer> res = CommonTasksSupport.rmFile(execEnv, dst, null);
    assertEquals("Error removing " + execEnv + ":" + dst, 0, res.get().intValue());
}
 

@Test
public void testProcess() throws Exception {

    final RateTracker tracker = new RateTracker(20);

    int numThreads = 5;
    ExecutorService threadPool = Executors.newFixedThreadPool(numThreads);

    final AtomicReference<RateLimitUtil> limiter = new AtomicReference<RateLimitUtil>(RateLimitUtil.create(100));

    final AtomicBoolean stop = new AtomicBoolean(false);

    // stats
    final AtomicInteger totalOps = new AtomicInteger(0);

    final CyclicBarrier barrier = new CyclicBarrier(numThreads + 1);
    final CountDownLatch latch = new CountDownLatch(numThreads);

    for (int i = 0; i < numThreads; i++) {

        threadPool.submit(new Callable<Void>() {

            @Override
            public Void call() throws Exception {

                barrier.await();
                while (!stop.get() && !Thread.currentThread().isInterrupted()) {
                    if (limiter.get().acquire()) {
                        tracker.trackRate(1);
                        totalOps.incrementAndGet();
                    }
                }
                latch.countDown();
                return null;
            }
        });
    }

    barrier.await();

    Thread.sleep(4000);
    System.out.println("Changing rate to 120");
    limiter.set(RateLimitUtil.create(120));

    Thread.sleep(4000);
    System.out.println("Changing rate to 80");
    limiter.set(RateLimitUtil.create(80));

    Thread.sleep(4000);
    System.out.println("Changing rate to 200");
    limiter.set(RateLimitUtil.create(200));

    Thread.sleep(4000);
    System.out.println("Changing rate to 100");
    limiter.set(RateLimitUtil.create(100));

    stop.set(true);
    threadPool.shutdownNow();

    //Thread.sleep(100);
    latch.await();

    System.out.println("=======================");
    System.out.println("Won lock: " + tracker.getWonLockCount());
    System.out.println("Total ops: " + totalOps.get());

    Assert.assertEquals(20, tracker.rWindow.getQueueSize());
    Assert.assertTrue(16 >= tracker.rWindow.getBucketCreateCount());

    List<Bucket> allBuckets = tracker.getAllBuckets();

    // Remove the first bucket since it's essentially unreliable since that is when the test had stopped.
    allBuckets.remove(0);

    for (Bucket b : allBuckets) {
        System.out.print(" " + b.count());
    }
    System.out.println("");
    Assert.assertTrue("P diff failed", 10 >= percentageDiff(200, allBuckets.get(0).count()));
    Assert.assertTrue("P diff failed", 10 >= percentageDiff(200, allBuckets.get(1).count()));
    Assert.assertTrue("P diff failed", 10 >= percentageDiff(200, allBuckets.get(2).count()));

    Assert.assertTrue("P diff failed", 10 >= percentageDiff(80, allBuckets.get(4).count()));
    Assert.assertTrue("P diff failed", 10 >= percentageDiff(80, allBuckets.get(5).count()));
    Assert.assertTrue("P diff failed", 10 >= percentageDiff(80, allBuckets.get(6).count()));

    Assert.assertTrue("P diff failed", 10 >= percentageDiff(120, allBuckets.get(8).count()));
    Assert.assertTrue("P diff failed", 10 >= percentageDiff(120, allBuckets.get(9).count()));
    Assert.assertTrue("P diff failed", 10 >= percentageDiff(120, allBuckets.get(10).count()));

    Assert.assertTrue("P diff failed", 10 >= percentageDiff(100, allBuckets.get(12).count()));
    Assert.assertTrue("P diff failed", 10 >= percentageDiff(100, allBuckets.get(13).count()));
    Assert.assertTrue("P diff failed", 10 >= percentageDiff(100, allBuckets.get(14).count()));
}
 

public void runSimplePublishTest(String subject, String replyTo, String bodyString) throws IOException, InterruptedException,ExecutionException {
    CompletableFuture<Boolean> gotPub = new CompletableFuture<>();
    AtomicReference<String> body  = new AtomicReference<>("");
    AtomicReference<String> protocol  = new AtomicReference<>("");

    NatsServerProtocolMock.Customizer receiveMessageCustomizer = (ts, r,w) -> {
        String pubLine = "";
        String bodyLine = "";
        
        System.out.println("*** Mock Server @" + ts.getPort() + " waiting for PUB ...");
        try {
            pubLine = r.readLine();
            bodyLine = r.readLine(); // Ignores encoding, but ok for test
        } catch(Exception e) {
            gotPub.cancel(true);
            return;
        }

        if (pubLine.startsWith("PUB")) {
            System.out.println("*** Mock Server @" + ts.getPort() + " got PUB ...");
            protocol.set(pubLine);
            body.set(bodyLine);
            gotPub.complete(Boolean.TRUE);
        }
    };

    try (NatsServerProtocolMock ts = new NatsServerProtocolMock(receiveMessageCustomizer);
                Connection  nc = Nats.connect(ts.getURI())) {
        byte[] bodyBytes = (bodyString != null) ? bodyString.getBytes(StandardCharsets.UTF_8) : null;

        assertTrue("Connected Status", Connection.Status.CONNECTED == nc.getStatus());

        nc.publish(subject, replyTo, bodyBytes);

        // This is used for the default test
        if (bodyString == null) {
            bodyBytes = new byte[0];
            bodyString = "";
        }

        assertTrue("Got pub.", gotPub.get().booleanValue()); //wait for receipt to close up
        nc.close();
        assertTrue("Closed Status", Connection.Status.CLOSED == nc.getStatus());

        String expectedProtocol = null;
        if (replyTo == null) {
            expectedProtocol = "PUB "+subject+" "+bodyBytes.length;
        } else {
            expectedProtocol = "PUB "+subject+" "+replyTo+" "+bodyBytes.length;
        }
        assertEquals("Protocol matches", expectedProtocol, protocol.get());

        assertEquals("Body matches", bodyString, body.get());
    }
}
 

private static void parentProcess() throws Throwable {
    JDKToolLauncher launcher = JDKToolLauncher
            .createUsingTestJDK("java")
            .addToolArg("MultiDead")
            .addToolArg("child");
    ProcessBuilder pb = new ProcessBuilder(launcher.getCommand());

    AtomicReference<Process> child = new AtomicReference<>();
    AtomicBoolean stopFlag = new AtomicBoolean(false);

    Thread th = new Thread(() -> {
        for (int i = 0; i < CHILDREN_COUNT; ++i) {
            System.out.println("child #" + (i + 1) + " of " +
                    CHILDREN_COUNT);
            long start = System.nanoTime();
            try {
                child.set(pb.start());
                child.get().waitFor();
                if (stopFlag.get()) {
                    break;
                }
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
            if (System.nanoTime() - start >
                    MILLISECONDS.toNanos(CHILD_TIMEOUT)) {
                System.err.println("Machine is too slow, " +
                        "skipping the test...");
                break;
            }
        }
    });

    th.start();
    th.join(TIMEOUT);

    stopFlag.set(true);
    if (th.isAlive()) {
        if (child.get() != null) {
            child.get().destroyForcibly();
        }
        throw new RuntimeException("Failed to complete on time.");
    }
}
 

public static <T> T await(Uni<T> upstream, Duration duration) {
    nonNull(upstream, "upstream");
    validate(duration);

    CountDownLatch latch = new CountDownLatch(1);
    AtomicReference<T> reference = new AtomicReference<>();
    AtomicReference<Throwable> referenceToFailure = new AtomicReference<>();
    UniSubscriber<T> subscriber = new UniSubscriber<T>() {
        @Override
        public void onSubscribe(UniSubscription subscription) {
            // Do nothing.
        }

        @Override
        public void onItem(T item) {
            reference.set(item);
            latch.countDown();
        }

        @Override
        public void onFailure(Throwable failure) {
            referenceToFailure.compareAndSet(null, failure);
            latch.countDown();
        }
    };
    AbstractUni.subscribe(upstream, subscriber);
    try {
        if (duration != null) {
            if (!latch.await(duration.toMillis(), TimeUnit.MILLISECONDS)) {
                referenceToFailure.compareAndSet(null, new TimeoutException());
            }
        } else {
            latch.await();
        }
    } catch (InterruptedException e) {
        referenceToFailure.compareAndSet(null, e);
        Thread.currentThread().interrupt();
    }

    Throwable throwable = referenceToFailure.get();
    if (throwable != null) {
        if (throwable instanceof RuntimeException) {
            throw (RuntimeException) throwable;
        }
        throw new CompletionException(throwable);
    } else {
        return reference.get();
    }
}
 

/**
 * Start a continuously executing set of duplex streaming ping-pong calls that will terminate when
 * {@code done.get()} is true. Each completed call will increment the counter by the specified
 * delta which benchmarks can use to measure messages per second or bandwidth.
 */
protected CountDownLatch startStreamingCalls(int callsPerChannel, final AtomicLong counter,
    final AtomicBoolean record, final AtomicBoolean done, final long counterDelta) {
  final CountDownLatch latch = new CountDownLatch(callsPerChannel * channels.length);
  for (final ManagedChannel channel : channels) {
    for (int i = 0; i < callsPerChannel; i++) {
      final ClientCall<ByteBuf, ByteBuf> streamingCall =
          channel.newCall(pingPongMethod, CALL_OPTIONS);
      final AtomicReference<StreamObserver<ByteBuf>> requestObserverRef =
          new AtomicReference<StreamObserver<ByteBuf>>();
      final AtomicBoolean ignoreMessages = new AtomicBoolean();
      StreamObserver<ByteBuf> requestObserver = ClientCalls.asyncBidiStreamingCall(
          streamingCall,
          new StreamObserver<ByteBuf>() {
            @Override
            public void onNext(ByteBuf value) {
              if (done.get()) {
                if (!ignoreMessages.getAndSet(true)) {
                  requestObserverRef.get().onCompleted();
                }
                return;
              }
              requestObserverRef.get().onNext(request.slice());
              if (record.get()) {
                counter.addAndGet(counterDelta);
              }
              // request is called automatically because the observer implicitly has auto
              // inbound flow control
            }

            @Override
            public void onError(Throwable t) {
              logger.log(Level.WARNING, "call error", t);
              latch.countDown();
            }

            @Override
            public void onCompleted() {
              latch.countDown();
            }
          });
      requestObserverRef.set(requestObserver);
      requestObserver.onNext(request.slice());
      requestObserver.onNext(request.slice());
    }
  }
  return latch;
}
 

@Override
public void addComplete(final int rc, LedgerHandle handle,
                        final long entryId, final Object ctx) {
    final AtomicReference<Integer> effectiveRC = new AtomicReference<Integer>(rc);
    try {
        if (FailpointUtils.checkFailPoint(FailpointUtils.FailPointName.FP_TransmitComplete)) {
            effectiveRC.set(BKException.Code.UnexpectedConditionException);
        }
    } catch (Exception exc) {
        effectiveRC.set(BKException.Code.UnexpectedConditionException);
    }

    // Sanity check to make sure we're receiving these callbacks in order.
    if (entryId > -1 && lastEntryId >= entryId) {
        LOG.error("Log segment {} saw out of order entry {} lastEntryId {}",
            new Object[] {fullyQualifiedLogSegment, entryId, lastEntryId});
    }
    lastEntryId = entryId;

    assert (ctx instanceof BKTransmitPacket);
    final BKTransmitPacket transmitPacket = (BKTransmitPacket) ctx;

    // Time from transmit until receipt of addComplete callback
    addCompleteTime.registerSuccessfulEvent(TimeUnit.MICROSECONDS.convert(
        System.nanoTime() - transmitPacket.getTransmitTime(), TimeUnit.NANOSECONDS));

    if (BKException.Code.OK == rc) {
        EntryBuffer recordSet = transmitPacket.getRecordSet();
        if (recordSet.hasUserRecords()) {
            synchronized (this) {
                lastTxIdAcknowledged = Math.max(lastTxIdAcknowledged, recordSet.getMaxTxId());
            }
        }
    }

    if (null != addCompleteFuturePool) {
        final Stopwatch queuedTime = Stopwatch.createStarted();
        addCompleteFuturePool.apply(new Function0<Void>() {
            public Void apply() {
                final Stopwatch deferredTime = Stopwatch.createStarted();
                addCompleteQueuedTime.registerSuccessfulEvent(queuedTime.elapsed(TimeUnit.MICROSECONDS));
                addCompleteDeferredProcessing(transmitPacket, entryId, effectiveRC.get());
                addCompleteDeferredTime.registerSuccessfulEvent(deferredTime.elapsed(TimeUnit.MICROSECONDS));
                return null;
            }
            @Override
            public String toString() {
                return String.format("AddComplete(Stream=%s, entryId=%d, rc=%d)",
                        fullyQualifiedLogSegment, entryId, rc);
            }
        }).addEventListener(new FutureEventListener<Void>() {
            @Override
            public void onSuccess(Void done) {
            }
            @Override
            public void onFailure(Throwable cause) {
                LOG.error("addComplete processing failed for {} entry {} lastTxId {} rc {} with error",
                    new Object[] {fullyQualifiedLogSegment, entryId, transmitPacket.getRecordSet().getMaxTxId(), rc, cause});
            }
        });
        // Race condition if we notify before the addComplete is enqueued.
        transmitPacket.notifyTransmitComplete(effectiveRC.get());
        outstandingTransmits.getAndDecrement();
    } else {
        // Notify transmit complete must be called before deferred processing in the
        // sync case since otherwise callbacks in deferred processing may deadlock.
        transmitPacket.notifyTransmitComplete(effectiveRC.get());
        outstandingTransmits.getAndDecrement();
        addCompleteDeferredProcessing(transmitPacket, entryId, effectiveRC.get());
    }
}
 

@Override
public void onCommand(User sender, GuildWrapper guild, TextChannel channel, Message message, String[] args, Member member) {
    if (!guild.getGuild().getSelfMember().hasPermission(Permission.MANAGE_ROLES)) {
        MessageUtils.sendErrorMessage("I can't lock the chat due to lack of permissions! " +
                "I need the `Manage Roles` permission", channel);
        return;
    }

    String reason = null;
    long time = -1;

    @Nonnull
    AtomicReference<TextChannel> tc = new AtomicReference<>(channel);
    if (args.length >= 1) {
        TextChannel tmp = ParseUtils.parseChannel(guild.getGuild(), args[0], false);
        if (tmp != null)
            tc.set(tmp);

        if (tmp == null || args.length >= 2) {
            Long l = GeneralUtils.parseTime(tmp == null ? args[0] : args[1]);
            if (l == null) {
                MessageUtils.sendErrorMessage("Invalid time format! Please use something like `1h10m`", channel);
                return;
            }
            time = l;
        }

        if (tmp != null && time > 0)
            reason = MessageUtils.getMessage(args, 2);
        else if ((tmp == null && time > 0) || (tmp != null && time == -1))
            reason = MessageUtils.getMessage(args, 1);
        else
            reason = MessageUtils.getMessage(args, 0);

        if (reason.isEmpty()) reason = null;
    }

    PermissionOverride everyoneOvr = tc.get().getPermissionOverride(guild.getGuild().getPublicRole());
    boolean locking = !everyoneOvr.getDenied().contains(Permission.MESSAGE_WRITE);

    EnumSet<Permission> perm = EnumSet.of(Permission.MESSAGE_WRITE);
    EnumSet<Permission> empty = EnumSet.noneOf(Permission.class);

    tc.get().getPermissionOverride(guild.getGuild().getPublicRole()).getManager()
            .deny(locking ? perm : empty)
            .clear(locking ? empty : perm)
            .reason(reason + "\nDone by: " + sender.getIdLong())
            .queue();
    tc.get().putPermissionOverride(guild.getGuild().getSelfMember())
            .setPermissions(locking ? perm : empty, empty)
            .reason(reason + "\nDone by: " + sender.getIdLong())
            .queue();

    if (tc.get().getIdLong() != channel.getIdLong())
        channel.sendMessage(new EmbedBuilder().setColor(locking ? ColorUtils.RED : ColorUtils.GREEN)
                .setDescription(tc.get().getAsMention() + " has been " + (locking ? "locked" : "unlocked") + "!")
                .build()).queue();

    if (guild.getGuild().getSelfMember().hasPermission(tc.get(), Permission.MESSAGE_WRITE))
        channel.sendMessage(new EmbedBuilder().setColor(locking ? ColorUtils.RED : ColorUtils.GREEN)
                .setDescription("The chat has been " + (locking ? "locked" : "unlocked") + " by a staff member"
                        + (locking && time > 0 ? " for "
                        + FormatUtils.formatTime(time, TimeUnit.MILLISECONDS, true, false) : "") + "!"
                        + (reason != null ? "\nReason: " + reason : ""))
                .build()).queue();

    if (locking && time > 0) {
        new FlareBotTask("ChannelUnlock-" + tc.get().getIdLong()) {
            @Override
            public void run() {
                tc.get().getPermissionOverride(guild.getGuild().getPublicRole()).getManager()
                        .clear(Permission.MESSAGE_WRITE)
                        .queue();

                if (guild.getGuild().getSelfMember().hasPermission(tc.get(), Permission.MESSAGE_WRITE))
                    channel.sendMessage(new EmbedBuilder().setColor(ColorUtils.GREEN)
                            .setDescription("The chat has been unlocked")
                            .build()).queue();
            }
        }.delay(time);
    }
}