text:
One of my favorite parts about my frontend JS workflow is using Live
Reload with Grunt/Gulp. Saving a few keystrokes a couple times a
minute is pretty great, especially if I can stay on the home
keys. Grunt and Gulp have a task that watches the files in your
project and then you kick can off certain tasks when they change, like
refreshing your display and re-running all the unit tests. In Scala,
you can get a similar thing by invoking your `sbt` command prefixed
with a `~`, but I didn't have anything for my Perl projects (not to
mention a general solution!).

At first, I looked for a similar tool that would watch files for
changes and let me hook into that action. I found
[App::Prove::Watch](https://metacpan.org/pod/App::Prove::Watch) which
was a Perl-only solution. It worked swimmingly after a quick PR, but
it only solved the problem for Perl projects. It wasn't until a few
days later that I realized I could leverage Emacs to get a reasonably
general solution, only depending on use of the `*compilation*` buffer
(no idea why it didn't occur to me earlier!).

Of course, Emacs knows exactly when my files change, as we always
eventually invoke `(save-buffer)` - even if it's advised with with
whitespace cleanup, or if it's invoked via some sort of
autosave. Instead of watching files on disk for changes, we can just
use Emacs to have the `(save-command)` trigger our tasks - aka
re-compiling the `*compilation*` buffer.

I used a global variable `ar-auto-recompile` to turn autocompilation
on and off, and made a function to handle the toggling for me:

    (setq ar-auto-recompile nil)
    (defun ar-auto-re-compile ()
      (interactive)
      (setq ar-auto-recompile (not ar-auto-recompile))
      (message (format "Auto recompile is now %s"
                       (if ar-auto-recompile "ON" "OFF"))))

It just toggles the variable and messages us about it. Next, our
advice to save-buffer should check whether the user wants
auto-recompilation. I also decided I only wanted it to re-compile if
the `*compilation*` buffer was in a visible window, which
`(get-buffer-window)` happily tells us.

    (defadvice save-buffer (after ar-auto-recompile activate)
      (when (and ar-auto-recompile
                 (get-buffer-window "*compilation*"))
        (set-buffer compilation-last-buffer)
        (revert-buffer t t)))

This solution isn't _quite_ the same as the watch behavior of the
other tools, as they smartly only respond to changes in files in their
own project. When turned on, this would blindly recompile any time a
buffer is saved and the `*compilation*` buffer is open. But, I don't
really mind - a few extra compilations probably won't hurt anything
unless the compilation wasn't safe to repeat in the first place.

So, my workflow is now to open up my test file, toggle on the
recompilation with `(ar-auto-re-compile)`, run a test to bring up the
`*compilation*` buffer. I keep that buffer visible and then do edits
and such as usual, [saving with `M-s`][save], which automatically
re-runs the test. And, if I need a more complicated compilation
command, it's just a `C-u M-x compile` away, and then that can get
repeated for me. As an extra bonus, this works perfectly with
[projectile][]'s compilation and test running commands (by default on
C-c p c and C-c p P), as they utilize the
`*compilation*` buffer as well. I also made a key-chord to switch the
current buffer over to the compilation one, as I often accidentally
close it:

    (key-chord-define-global "vv" (lambda () (interactive)
                                    (switch-to-buffer "*compilation*")))

Finally, if you don't want to use advice, it's [simple enough][] to
use a wrapper around `(save-buffer)` and update your save keybinding
to use your own function. And as usual, the whole snippet is in my
github [somewhere][].

[save]:
https://github.com/gempesaw/dotemacs/blob/6ca8a3995d558ac924e576bd60c516dbd1c450e2/dg-elisp/dg-kbd.el#L135-L140
[projectile]: https://github.com/bbatsov/projectile
[simple enough]: https://github.com/gempesaw/dotemacs/blob/8e7cb17e72339069d5e318fd7eb44d4718faa36c/dg-elisp/dg-defun.el#L461-L466
[somewhere]: https://github.com/gempesaw/dotemacs/blob/6ca8a3995d558ac924e576bd60c516dbd1c450e2/dg-elisp/dg-defun.el#L459-L470


text:
package org.apache.pulsar.client.impl;

import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkState;
import static org.apache.pulsar.client.impl.TransactionMetaStoreHandler.getExceptionByServerError;
import static org.apache.pulsar.common.util.Runnables.catchingAndLoggingThrowables;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Queues;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.unix.Errors.NativeIoException;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.util.concurrent.Promise;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.net.URISyntaxException;
import java.nio.channels.ClosedChannelException;
import java.util.Arrays;
import java.util.List;
import java.util.Optional;
import java.util.Queue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import lombok.AccessLevel;
import lombok.Getter;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.pulsar.PulsarVersion;
import org.apache.pulsar.client.api.Authentication;
import org.apache.pulsar.client.api.AuthenticationDataProvider;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.PulsarClientException.ConnectException;
import org.apache.pulsar.client.api.PulsarClientException.TimeoutException;
import org.apache.pulsar.client.impl.BinaryProtoLookupService.LookupDataResult;
import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
import org.apache.pulsar.client.impl.schema.SchemaInfoUtil;
import org.apache.pulsar.client.impl.transaction.TransactionBufferHandler;
import org.apache.pulsar.client.util.TimedCompletableFuture;
import org.apache.pulsar.common.api.AuthData;
import org.apache.pulsar.common.api.proto.BaseCommand;
import org.apache.pulsar.common.api.proto.CommandAckResponse;
import org.apache.pulsar.common.api.proto.CommandActiveConsumerChange;
import org.apache.pulsar.common.api.proto.CommandAddPartitionToTxnResponse;
import org.apache.pulsar.common.api.proto.CommandAddSubscriptionToTxnResponse;
import org.apache.pulsar.common.api.proto.CommandAuthChallenge;
import org.apache.pulsar.common.api.proto.CommandCloseConsumer;
import org.apache.pulsar.common.api.proto.CommandCloseProducer;
import org.apache.pulsar.common.api.proto.CommandConnected;
import org.apache.pulsar.common.api.proto.CommandEndTxnOnPartitionResponse;
import org.apache.pulsar.common.api.proto.CommandEndTxnOnSubscriptionResponse;
import org.apache.pulsar.common.api.proto.CommandEndTxnResponse;
import org.apache.pulsar.common.api.proto.CommandError;
import org.apache.pulsar.common.api.proto.CommandGetLastMessageIdResponse;
import org.apache.pulsar.common.api.proto.CommandGetOrCreateSchemaResponse;
import org.apache.pulsar.common.api.proto.CommandGetSchemaResponse;
import org.apache.pulsar.common.api.proto.CommandGetTopicsOfNamespaceResponse;
import org.apache.pulsar.common.api.proto.CommandLookupTopicResponse;
import org.apache.pulsar.common.api.proto.CommandMessage;
import org.apache.pulsar.common.api.proto.CommandNewTxnResponse;
import org.apache.pulsar.common.api.proto.CommandPartitionedTopicMetadataResponse;
import org.apache.pulsar.common.api.proto.CommandProducerSuccess;
import org.apache.pulsar.common.api.proto.CommandReachedEndOfTopic;
import org.apache.pulsar.common.api.proto.CommandSendError;
import org.apache.pulsar.common.api.proto.CommandSendReceipt;
import org.apache.pulsar.common.api.proto.CommandSuccess;
import org.apache.pulsar.common.api.proto.CommandTcClientConnectResponse;
import org.apache.pulsar.common.api.proto.CommandTopicMigrated;
import org.apache.pulsar.common.api.proto.CommandTopicMigrated.ResourceType;
import org.apache.pulsar.common.api.proto.CommandWatchTopicListSuccess;
import org.apache.pulsar.common.api.proto.CommandWatchTopicUpdate;
import org.apache.pulsar.common.api.proto.ServerError;
import org.apache.pulsar.common.lookup.GetTopicsResult;
import org.apache.pulsar.common.protocol.Commands;
import org.apache.pulsar.common.protocol.PulsarHandler;
import org.apache.pulsar.common.protocol.schema.SchemaVersion;
import org.apache.pulsar.common.schema.SchemaInfo;
import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.common.util.collections.ConcurrentLongHashMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@SuppressWarnings("unchecked")
public class ClientCnx extends PulsarHandler {

    protected final Authentication authentication;
    protected State state;

    @Getter
    private final ConcurrentLongHashMap<TimedCompletableFuture<? extends Object>> pendingRequests =
            ConcurrentLongHashMap.<TimedCompletableFuture<? extends Object>>newBuilder()
                    .expectedItems(16)
                    .concurrencyLevel(1)
                    .build();
    // LookupRequests that waiting in client side.
    private final Queue<Pair<Long, Pair<ByteBuf, TimedCompletableFuture<LookupDataResult>>>> waitingLookupRequests;

    @VisibleForTesting
    final ConcurrentLongHashMap<ProducerImpl<?>> producers =
            ConcurrentLongHashMap.<ProducerImpl<?>>newBuilder()
                    .expectedItems(16)
                    .concurrencyLevel(1)
                    .build();
    @VisibleForTesting
    final ConcurrentLongHashMap<ConsumerImpl<?>> consumers =
            ConcurrentLongHashMap.<ConsumerImpl<?>>newBuilder()
                    .expectedItems(16)
                    .concurrencyLevel(1)
                    .build();
    private final ConcurrentLongHashMap<TransactionMetaStoreHandler> transactionMetaStoreHandlers =
            ConcurrentLongHashMap.<TransactionMetaStoreHandler>newBuilder()
                    .expectedItems(16)
                    .concurrencyLevel(1)
                    .build();
    @Getter(AccessLevel.PACKAGE)
    private final ConcurrentLongHashMap<TopicListWatcher> topicListWatchers =
            ConcurrentLongHashMap.<TopicListWatcher>newBuilder()
                    .expectedItems(16)
                    .concurrencyLevel(1)
                    .build();

    private final CompletableFuture<Void> connectionFuture = new CompletableFuture<Void>();
    private final ConcurrentLinkedQueue<RequestTime> requestTimeoutQueue = new ConcurrentLinkedQueue<>();

    @VisibleForTesting
    @Getter(AccessLevel.PACKAGE)
    private final Semaphore pendingLookupRequestSemaphore;
    private final Semaphore maxLookupRequestSemaphore;
    private final EventLoopGroup eventLoopGroup;

    private static final AtomicIntegerFieldUpdater<ClientCnx> NUMBER_OF_REJECTED_REQUESTS_UPDATER =
            AtomicIntegerFieldUpdater.newUpdater(ClientCnx.class, "numberOfRejectRequests");
    @SuppressWarnings("unused")
    private volatile int numberOfRejectRequests = 0;

    @Getter
    private static int maxMessageSize = Commands.DEFAULT_MAX_MESSAGE_SIZE;

    private final int maxNumberOfRejectedRequestPerConnection;
    private final int rejectedRequestResetTimeSec = 60;
    protected final int protocolVersion;
    private final long operationTimeoutMs;

    protected String proxyToTargetBrokerAddress = null;
    // Remote hostName with which client is connected
    protected String remoteHostName = null;

    private ScheduledFuture<?> timeoutTask;
    private SocketAddress localAddress;
    private SocketAddress remoteAddress;

    // Added for mutual authentication.
    @Getter
    protected AuthenticationDataProvider authenticationDataProvider;
    private TransactionBufferHandler transactionBufferHandler;
    private boolean supportsTopicWatchers;

    /** Idle stat. **/
    @Getter
    private final ClientCnxIdleState idleState;

    @Getter
    private long lastDisconnectedTimestamp;

    protected enum State {
        None, SentConnectFrame, Ready, Failed, Connecting
    }

    private static class RequestTime {
        private final long creationTimeNanos;
        final long requestId;
        final RequestType requestType;

        RequestTime(long requestId, RequestType requestType) {
            this.creationTimeNanos = System.nanoTime();
            this.requestId = requestId;
            this.requestType = requestType;
        }

        boolean isTimedOut(long timeoutMillis) {
            long requestAgeMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - creationTimeNanos);
            return requestAgeMillis > timeoutMillis;
        }
    }

    private enum RequestType {
        Command,
        GetLastMessageId,
        GetTopics,
        GetSchema,
        GetOrCreateSchema,
        AckResponse,
        Lookup;

        String getDescription() {
            if (this == Command) {
                return "request";
            } else {
                return name() + " request";
            }
        }
    }

    public ClientCnx(ClientConfigurationData conf, EventLoopGroup eventLoopGroup) {
        this(conf, eventLoopGroup, Commands.getCurrentProtocolVersion());
    }

    public ClientCnx(ClientConfigurationData conf, EventLoopGroup eventLoopGroup, int protocolVersion) {
        super(conf.getKeepAliveIntervalSeconds(), TimeUnit.SECONDS);
        checkArgument(conf.getMaxLookupRequest() > conf.getConcurrentLookupRequest());
        this.pendingLookupRequestSemaphore = new Semaphore(conf.getConcurrentLookupRequest(), false);
        this.maxLookupRequestSemaphore =
                new Semaphore(conf.getMaxLookupRequest() - conf.getConcurrentLookupRequest(), false);
        this.waitingLookupRequests = Queues.newConcurrentLinkedQueue();
        this.authentication = conf.getAuthentication();
        this.eventLoopGroup = eventLoopGroup;
        this.maxNumberOfRejectedRequestPerConnection = conf.getMaxNumberOfRejectedRequestPerConnection();
        this.operationTimeoutMs = conf.getOperationTimeoutMs();
        this.state = State.None;
        this.protocolVersion = protocolVersion;
        this.idleState = new ClientCnxIdleState(this);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
        this.localAddress = ctx.channel().localAddress();
        this.remoteAddress = ctx.channel().remoteAddress();

        this.timeoutTask = this.eventLoopGroup
                .scheduleAtFixedRate(catchingAndLoggingThrowables(this::checkRequestTimeout), operationTimeoutMs,
                        operationTimeoutMs, TimeUnit.MILLISECONDS);

        if (proxyToTargetBrokerAddress == null) {
            if (log.isDebugEnabled()) {
                log.debug("{} Connected to broker", ctx.channel());
            }
        } else {
            log.info("{} Connected through proxy to target broker at {}", ctx.channel(), proxyToTargetBrokerAddress);
        }
        // Send CONNECT command
        ctx.writeAndFlush(newConnectCommand())
                .addListener(future -> {
                    if (future.isSuccess()) {
                        if (log.isDebugEnabled()) {
                            log.debug("Complete: {}", future.isSuccess());
                        }
                        state = State.SentConnectFrame;
                    } else {
                        log.warn("Error during handshake", future.cause());
                        ctx.close();
                    }
                });
    }

    protected ByteBuf newConnectCommand() throws Exception {
        // mutual authentication is to auth between `remoteHostName` and this client for this channel.
        // each channel will have a mutual client/server pair, mutual client evaluateChallenge with init data,
        // and return authData to server.
        authenticationDataProvider = authentication.getAuthData(remoteHostName);
        AuthData authData = authenticationDataProvider.authenticate(AuthData.INIT_AUTH_DATA);
        return Commands.newConnect(authentication.getAuthMethodName(), authData, this.protocolVersion,
                PulsarVersion.getVersion(), proxyToTargetBrokerAddress, null, null, null);
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        super.channelInactive(ctx);
        lastDisconnectedTimestamp = System.currentTimeMillis();
        log.info("{} Disconnected", ctx.channel());
        if (!connectionFuture.isDone()) {
            connectionFuture.completeExceptionally(new PulsarClientException("Connection already closed"));
        }

        ConnectException e = new ConnectException(
                "Disconnected from server at " + ctx.channel().remoteAddress());

        // Fail out all the pending ops
        pendingRequests.forEach((key, future) -> {
            if (pendingRequests.remove(key, future) && !future.isDone()) {
                future.completeExceptionally(e);
            }
        });
        waitingLookupRequests.forEach(pair -> pair.getRight().getRight().completeExceptionally(e));

        // Notify all attached producers/consumers so they have a chance to reconnect
        producers.forEach((id, producer) -> producer.connectionClosed(this));
        consumers.forEach((id, consumer) -> consumer.connectionClosed(this));
        transactionMetaStoreHandlers.forEach((id, handler) -> handler.connectionClosed(this));
        topicListWatchers.forEach((__, watcher) -> watcher.connectionClosed(this));

        waitingLookupRequests.clear();

        producers.clear();
        consumers.clear();
        topicListWatchers.clear();

        timeoutTask.cancel(true);
    }

    // Command Handlers

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        if (state != State.Failed) {
            // No need to report stack trace for known exceptions that happen in disconnections
            log.warn("[{}] Got exception {}", remoteAddress,
                    ClientCnx.isKnownException(cause) ? cause : ExceptionUtils.getStackTrace(cause));
            state = State.Failed;
        } else {
            // At default info level, suppress all subsequent exceptions that are thrown when the connection has already
            // failed
            if (log.isDebugEnabled()) {
                log.debug("[{}] Got exception: {}", remoteAddress, cause.getMessage(), cause);
            }
        }

        ctx.close();
    }

    public static boolean isKnownException(Throwable t) {
        return t instanceof NativeIoException || t instanceof ClosedChannelException;
    }

    @Override
    protected void handleConnected(CommandConnected connected) {
        checkArgument(state == State.SentConnectFrame || state == State.Connecting);
        if (connected.hasMaxMessageSize()) {
            if (log.isDebugEnabled()) {
                log.debug("{} Connection has max message size setting, replace old frameDecoder with "
                          + "server frame size {}", ctx.channel(), connected.getMaxMessageSize());
            }
            maxMessageSize = connected.getMaxMessageSize();
            ctx.pipeline().replace("frameDecoder", "newFrameDecoder", new LengthFieldBasedFrameDecoder(
                connected.getMaxMessageSize() + Commands.MESSAGE_SIZE_FRAME_PADDING, 0, 4, 0, 4));
        }
        if (log.isDebugEnabled()) {
            log.debug("{} Connection is ready", ctx.channel());
        }

        supportsTopicWatchers =
            connected.hasFeatureFlags() && connected.getFeatureFlags().isSupportsTopicWatchers();

        // set remote protocol version to the correct version before we complete the connection future
        setRemoteEndpointProtocolVersion(connected.getProtocolVersion());
        connectionFuture.complete(null);
        state = State.Ready;
    }

    @Override
    protected void handleAuthChallenge(CommandAuthChallenge authChallenge) {
        checkArgument(authChallenge.hasChallenge());
        checkArgument(authChallenge.getChallenge().hasAuthData());

        if (Arrays.equals(AuthData.REFRESH_AUTH_DATA_BYTES, authChallenge.getChallenge().getAuthData())) {
            try {
                authenticationDataProvider = authentication.getAuthData(remoteHostName);
            } catch (PulsarClientException e) {
                log.error("{} Error when refreshing authentication data provider: {}", ctx.channel(), e);
                connectionFuture.completeExceptionally(e);
                return;
            }
        }

        // mutual authn. If auth not complete, continue auth; if auth complete, complete connectionFuture.
        try {
            AuthData authData = authenticationDataProvider
                .authenticate(AuthData.of(authChallenge.getChallenge().getAuthData()));

            checkState(!authData.isComplete());

            ByteBuf request = Commands.newAuthResponse(authentication.getAuthMethodName(),
                authData,
                this.protocolVersion,
                PulsarVersion.getVersion());

            if (log.isDebugEnabled()) {
                log.debug("{} Mutual auth {}", ctx.channel(), authentication.getAuthMethodName());
            }

            ctx.writeAndFlush(request).addListener(writeFuture -> {
                if (!writeFuture.isSuccess()) {
                    log.warn("{} Failed to send request for mutual auth to broker: {}", ctx.channel(),
                        writeFuture.cause().getMessage());
                    connectionFuture.completeExceptionally(writeFuture.cause());
                }
            });

            if (state == State.SentConnectFrame) {
                state = State.Connecting;
            }
        } catch (Exception e) {
            log.error("{} Error mutual verify: {}", ctx.channel(), e);
            connectionFuture.completeExceptionally(e);
            return;
        }
    }

    @Override
    protected void handleSendReceipt(CommandSendReceipt sendReceipt) {
        checkArgument(state == State.Ready);

        long producerId = sendReceipt.getProducerId();
        long sequenceId = sendReceipt.getSequenceId();
        long highestSequenceId = sendReceipt.getHighestSequenceId();
        long ledgerId = -1;
        long entryId = -1;
        if (sendReceipt.hasMessageId()) {
            ledgerId = sendReceipt.getMessageId().getLedgerId();
            entryId = sendReceipt.getMessageId().getEntryId();
        }

        if (ledgerId == -1 && entryId == -1) {
            log.warn("{} Message with sequence-id {} published by producer {} has been dropped", ctx.channel(),
                    sequenceId, producerId);
        }

        if (log.isDebugEnabled()) {
            log.debug("{} Got receipt for producer: {} -- msg: {} -- id: {}:{}", ctx.channel(), producerId, sequenceId,
                    ledgerId, entryId);
        }

        ProducerImpl<?> producer = producers.get(producerId);
        if (producer != null) {
            producer.ackReceived(this, sequenceId, highestSequenceId, ledgerId, entryId);
        } else {
            if (log.isDebugEnabled()) {
                log.debug("Producer is {} already closed, ignore published message [{}-{}]", producerId, ledgerId,
                        entryId);
            }
        }
    }

    @Override
    protected void handleAckResponse(CommandAckResponse ackResponse) {
        checkArgument(state == State.Ready);
        checkArgument(ackResponse.getRequestId() >= 0);
        CompletableFuture<?> completableFuture = pendingRequests.remove(ackResponse.getRequestId());
        if (completableFuture != null && !completableFuture.isDone()) {
            if (!ackResponse.hasError()) {
                completableFuture.complete(null);
            } else {
                completableFuture.completeExceptionally(
                        getPulsarClientException(ackResponse.getError(),
                                                 buildError(ackResponse.getRequestId(), ackResponse.getMessage())));
            }
        } else {
            log.warn("AckResponse has complete when receive response! requestId : {}, consumerId : {}",
                    ackResponse.getRequestId(), ackResponse.hasConsumerId());
        }
    }

    @Override
    protected void handleMessage(CommandMessage cmdMessage, ByteBuf headersAndPayload) {
        checkArgument(state == State.Ready);

        if (log.isDebugEnabled()) {
            log.debug("{} Received a message from the server: {}", ctx.channel(), cmdMessage);
        }
        ConsumerImpl<?> consumer = consumers.get(cmdMessage.getConsumerId());
        if (consumer != null) {
            consumer.messageReceived(cmdMessage, headersAndPayload, this);
        }
    }

    @Override
    protected void handleActiveConsumerChange(CommandActiveConsumerChange change) {
        checkArgument(state == State.Ready);

        if (log.isDebugEnabled()) {
            log.debug("{} Received a consumer group change message from the server : {}", ctx.channel(), change);
        }
        ConsumerImpl<?> consumer = consumers.get(change.getConsumerId());
        if (consumer != null) {
            consumer.activeConsumerChanged(change.isIsActive());
        }
    }

    @Override
    protected void handleSuccess(CommandSuccess success) {
        checkArgument(state == State.Ready);

        if (log.isDebugEnabled()) {
            log.debug("{} Received success response from server: {}", ctx.channel(), success.getRequestId());
        }
        long requestId = success.getRequestId();
        CompletableFuture<?> requestFuture = pendingRequests.remove(requestId);
        if (requestFuture != null) {
            requestFuture.complete(null);
        } else {
            log.warn("{} Received unknown request id from server: {}", ctx.channel(), success.getRequestId());
        }
    }

    @Override
    protected void handleGetLastMessageIdSuccess(CommandGetLastMessageIdResponse success) {
        checkArgument(state == State.Ready);

        if (log.isDebugEnabled()) {
            log.debug("{} Received success GetLastMessageId response from server: {}",
                    ctx.channel(), success.getRequestId());
        }
        long requestId = success.getRequestId();
        CompletableFuture<CommandGetLastMessageIdResponse> requestFuture =
                (CompletableFuture<CommandGetLastMessageIdResponse>) pendingRequests.remove(requestId);
        if (requestFuture != null) {
            requestFuture.complete(new CommandGetLastMessageIdResponse().copyFrom(success));
        } else {
            log.warn("{} Received unknown request id from server: {}", ctx.channel(), success.getRequestId());
        }
    }

    @Override
    protected void handleProducerSuccess(CommandProducerSuccess success) {
        checkArgument(state == State.Ready);

        if (log.isDebugEnabled()) {
            log.debug("{} Received producer success response from server: {} - producer-name: {}", ctx.channel(),
                    success.getRequestId(), success.getProducerName());
        }
        long requestId = success.getRequestId();
        if (!success.isProducerReady()) {
            // We got a success operation but the producer is not ready. This means that the producer has been queued up
            // in broker. We need to leave the future pending until we get the final confirmation. We just mark that
            // we have received a response, in order to avoid the timeout.
            TimedCompletableFuture<?> requestFuture = pendingRequests.get(requestId);
            if (requestFuture != null) {
                log.info("{} Producer {} has been queued up at broker. request: {}", ctx.channel(),
                        success.getProducerName(), requestId);
                requestFuture.markAsResponded();
            }
            return;
        }

        CompletableFuture<ProducerResponse> requestFuture =
                (CompletableFuture<ProducerResponse>) pendingRequests.remove(requestId);
        if (requestFuture != null) {
            ProducerResponse pr = new ProducerResponse(success.getProducerName(),
                    success.getLastSequenceId(),
                    success.getSchemaVersion(),
                    success.hasTopicEpoch() ? Optional.of(success.getTopicEpoch()) : Optional.empty());
            requestFuture.complete(pr);
        } else {
            log.warn("{} Received unknown request id from server: {}", ctx.channel(), success.getRequestId());
        }
    }

    @Override
    protected void handleLookupResponse(CommandLookupTopicResponse lookupResult) {
        if (log.isDebugEnabled()) {
            CommandLookupTopicResponse.LookupType response =
                    lookupResult.hasResponse() ? lookupResult.getResponse() : null;
            log.debug("Received Broker lookup response: {} {}", lookupResult.getRequestId(), response);
        }

        long requestId = lookupResult.getRequestId();
        CompletableFuture<LookupDataResult> requestFuture = getAndRemovePendingLookupRequest(requestId);

        if (requestFuture != null) {
            if (requestFuture.isCompletedExceptionally()) {
                if (log.isDebugEnabled()) {
                    log.debug("{} Request {} already timed-out", ctx.channel(), lookupResult.getRequestId());
                }
                return;
            }
            // Complete future with exception if : Result.response=fail/null
            if (!lookupResult.hasResponse()
                    || CommandLookupTopicResponse.LookupType.Failed.equals(lookupResult.getResponse())) {
                if (lookupResult.hasError()) {
                    checkServerError(lookupResult.getError(), lookupResult.getMessage());
                    requestFuture.completeExceptionally(
                            getPulsarClientException(lookupResult.getError(),
                                    buildError(lookupResult.getRequestId(), lookupResult.getMessage())));
                } else {
                    requestFuture
                            .completeExceptionally(new PulsarClientException.LookupException("Empty lookup response"));
                }
            } else {
                requestFuture.complete(new LookupDataResult(lookupResult));
            }
        } else {
            log.warn("{} Received unknown request id from server: {}", ctx.channel(), lookupResult.getRequestId());
        }
    }

    @Override
    protected void handlePartitionResponse(CommandPartitionedTopicMetadataResponse lookupResult) {
        if (log.isDebugEnabled()) {
            CommandPartitionedTopicMetadataResponse.LookupType response =
                    lookupResult.hasResponse() ? lookupResult.getResponse() : null;
            int partitions = lookupResult.hasPartitions() ? lookupResult.getPartitions() : -1;
            log.debug("Received Broker Partition response: {} {} {}", lookupResult.getRequestId(), response,
                    partitions);
        }

        long requestId = lookupResult.getRequestId();
        CompletableFuture<LookupDataResult> requestFuture = getAndRemovePendingLookupRequest(requestId);

        if (requestFuture != null) {
            if (requestFuture.isCompletedExceptionally()) {
                if (log.isDebugEnabled()) {
                    log.debug("{} Request {} already timed-out", ctx.channel(), lookupResult.getRequestId());
                }
                return;
            }
            // Complete future with exception if : Result.response=fail/null
            if (!lookupResult.hasResponse()
                    || CommandPartitionedTopicMetadataResponse.LookupType.Failed.equals(lookupResult.getResponse())) {
                if (lookupResult.hasError()) {
                    String message = buildError(lookupResult.getRequestId(),
                                                lookupResult.hasMessage() ? lookupResult.getMessage() : null);
                    checkServerError(lookupResult.getError(), message);
                    requestFuture.completeExceptionally(
                            getPulsarClientException(lookupResult.getError(), message));
                } else {
                    requestFuture
                            .completeExceptionally(new PulsarClientException.LookupException("Empty lookup response"));
                }
            } else {
                // return LookupDataResult when Result.response = success/redirect
                requestFuture.complete(new LookupDataResult(lookupResult.getPartitions()));
            }
        } else {
            log.warn("{} Received unknown request id from server: {}", ctx.channel(), lookupResult.getRequestId());
        }
    }

    @Override
    protected void handleReachedEndOfTopic(CommandReachedEndOfTopic commandReachedEndOfTopic) {
        final long consumerId = commandReachedEndOfTopic.getConsumerId();

        log.info("[{}] Broker notification reached the end of topic: {}", remoteAddress, consumerId);

        ConsumerImpl<?> consumer = consumers.get(consumerId);
        if (consumer != null) {
            consumer.setTerminated();
        }
    }

    @Override
    protected void handleTopicMigrated(CommandTopicMigrated commandTopicMigrated) {
        final long resourceId = commandTopicMigrated.getResourceId();
        final String serviceUrl = commandTopicMigrated.getBrokerServiceUrl();
        final String serviceUrlTls = commandTopicMigrated.getBrokerServiceUrlTls();

        HandlerState resource = commandTopicMigrated.getResourceType() == ResourceType.Producer
                ? producers.get(resourceId)
                : consumers.get(resourceId);
        log.info("{} is migrated to {}/{}", commandTopicMigrated.getResourceType().name(), serviceUrl, serviceUrlTls);
        if (resource != null) {
            try {
                resource.setRedirectedClusterURI(serviceUrl, serviceUrlTls);
            } catch (URISyntaxException e) {
                log.info("[{}] Invalid redirect url {}/{} for {}", remoteAddress, serviceUrl, serviceUrlTls,
                        resourceId);
            }
        }
    }

    // caller of this method needs to be protected under pendingLookupRequestSemaphore
    private void addPendingLookupRequests(long requestId, TimedCompletableFuture<LookupDataResult> future) {
        pendingRequests.put(requestId, future);
        requestTimeoutQueue.add(new RequestTime(requestId, RequestType.Lookup));
    }

    private CompletableFuture<LookupDataResult> getAndRemovePendingLookupRequest(long requestId) {
        CompletableFuture<LookupDataResult> result =
                (CompletableFuture<LookupDataResult>) pendingRequests.remove(requestId);
        if (result != null) {
            Pair<Long, Pair<ByteBuf, TimedCompletableFuture<LookupDataResult>>> firstOneWaiting =
                    waitingLookupRequests.poll();
            if (firstOneWaiting != null) {
                maxLookupRequestSemaphore.release();
                // schedule a new lookup in.
                eventLoopGroup.execute(() -> {
                    long newId = firstOneWaiting.getLeft();
                    TimedCompletableFuture<LookupDataResult> newFuture = firstOneWaiting.getRight().getRight();
                    addPendingLookupRequests(newId, newFuture);
                    ctx.writeAndFlush(firstOneWaiting.getRight().getLeft()).addListener(writeFuture -> {
                        if (!writeFuture.isSuccess()) {
                            log.warn("{} Failed to send request {} to broker: {}", ctx.channel(), newId,
                                writeFuture.cause().getMessage());
                            getAndRemovePendingLookupRequest(newId);
                            newFuture.completeExceptionally(writeFuture.cause());
                        }
                    });
                });
            } else {
                pendingLookupRequestSemaphore.release();
            }
        }
        return result;
    }

    @Override
    protected void handleSendError(CommandSendError sendError) {
        log.warn("{} Received send error from server: {} : {}", ctx.channel(), sendError.getError(),
                sendError.getMessage());

        long producerId = sendError.getProducerId();
        long sequenceId = sendError.getSequenceId();

        switch (sendError.getError()) {
        case ChecksumError:
            producers.get(producerId).recoverChecksumError(this, sequenceId);
            break;

        case TopicTerminatedError:
            producers.get(producerId).terminated(this);
            break;
        case NotAllowedError:
            producers.get(producerId).recoverNotAllowedError(sequenceId, sendError.getMessage());
            break;

        default:
            // By default, for transient error, let the reconnection logic
            // to take place and re-establish the produce again
            ctx.close();
        }
    }

    @Override
    protected void handleError(CommandError error) {
        checkArgument(state == State.SentConnectFrame || state == State.Ready);

        log.warn("{} Received error from server: {}", ctx.channel(), error.getMessage());
        long requestId = error.getRequestId();
        if (error.getError() == ServerError.ProducerBlockedQuotaExceededError) {
            log.warn("{} Producer creation has been blocked because backlog quota exceeded for producer topic",
                    ctx.channel());
        }
        if (error.getError() == ServerError.AuthenticationError) {
            connectionFuture.completeExceptionally(
                    new PulsarClientException.AuthenticationException(error.getMessage()));
            log.error("{} Failed to authenticate the client", ctx.channel());
        }
        if (error.getError() == ServerError.NotAllowedError) {
            log.error("Get not allowed error, {}", error.getMessage());
            connectionFuture.completeExceptionally(new PulsarClientException.NotAllowedException(error.getMessage()));
        }
        CompletableFuture<?> requestFuture = pendingRequests.remove(requestId);
        if (requestFuture != null) {
            requestFuture.completeExceptionally(
                    getPulsarClientException(error.getError(),
                                             buildError(error.getRequestId(), error.getMessage())));
        } else {
            log.warn("{} Received unknown request id from server: {}", ctx.channel(), error.getRequestId());
        }
    }

    @Override
    protected void handleCloseProducer(CommandCloseProducer closeProducer) {
        log.info("[{}] Broker notification of Closed producer: {}", remoteAddress, closeProducer.getProducerId());
        final long producerId = closeProducer.getProducerId();
        ProducerImpl<?> producer = producers.remove(producerId);
        if (producer != null) {
            producer.connectionClosed(this);
        } else {
            log.warn("Producer with id {} not found while closing producer ", producerId);
        }
    }

    @Override
    protected void handleCloseConsumer(CommandCloseConsumer closeConsumer) {
        log.info("[{}] Broker notification of Closed consumer: {}", remoteAddress, closeConsumer.getConsumerId());
        final long consumerId = closeConsumer.getConsumerId();
        ConsumerImpl<?> consumer = consumers.remove(consumerId);
        if (consumer != null) {
            consumer.connectionClosed(this);
        } else {
            log.warn("Consumer with id {} not found while closing consumer ", consumerId);
        }
    }

    @Override
    protected boolean isHandshakeCompleted() {
        return state == State.Ready;
    }

    public CompletableFuture<LookupDataResult> newLookup(ByteBuf request, long requestId) {
        TimedCompletableFuture<LookupDataResult> future = new TimedCompletableFuture<>();

        if (pendingLookupRequestSemaphore.tryAcquire()) {
            future.whenComplete((lookupDataResult, throwable) -> {
                if (throwable instanceof ConnectException) {
                    pendingLookupRequestSemaphore.release();
                }
            });
            addPendingLookupRequests(requestId, future);
            ctx.writeAndFlush(request).addListener(writeFuture -> {
                if (!writeFuture.isSuccess()) {
                    log.warn("{} Failed to send request {} to broker: {}", ctx.channel(), requestId,
                        writeFuture.cause().getMessage());
                    getAndRemovePendingLookupRequest(requestId);
                    future.completeExceptionally(writeFuture.cause());
                }
            });
        } else {
            if (log.isDebugEnabled()) {
                log.debug("{} Failed to add lookup-request into pending queue", requestId);
            }

            if (maxLookupRequestSemaphore.tryAcquire()) {
                waitingLookupRequests.add(Pair.of(requestId, Pair.of(request, future)));
            } else {
                if (log.isDebugEnabled()) {
                    log.debug("{} Failed to add lookup-request into waiting queue", requestId);
                }
                future.completeExceptionally(new PulsarClientException.TooManyRequestsException(String.format(
                    "Requests number out of config: There are {%s} lookup requests outstanding and {%s} requests"
                            + " pending.",
                    pendingLookupRequestSemaphore.getQueueLength(),
                    waitingLookupRequests.size())));
            }
        }
        return future;
    }

    public CompletableFuture<GetTopicsResult> newGetTopicsOfNamespace(ByteBuf request, long requestId) {
        return sendRequestAndHandleTimeout(request, requestId, RequestType.GetTopics, true);
    }

    public CompletableFuture<Void> newAckForReceipt(ByteBuf request, long requestId) {
        return sendRequestAndHandleTimeout(request, requestId, RequestType.AckResponse, true);
    }

    public void newAckForReceiptWithFuture(ByteBuf request, long requestId,
                                           TimedCompletableFuture<Void> future) {
        sendRequestAndHandleTimeout(request, requestId, RequestType.AckResponse, false, future);
    }

    @Override
    protected void handleGetTopicsOfNamespaceSuccess(CommandGetTopicsOfNamespaceResponse success) {
        checkArgument(state == State.Ready);

        long requestId = success.getRequestId();
        List<String> topics = success.getTopicsList();

        if (log.isDebugEnabled()) {
            log.debug("{} Received get topics of namespace success response from server: {} - topics.size: {}",
                ctx.channel(), success.getRequestId(), topics.size());
        }

        CompletableFuture<GetTopicsResult> requestFuture =
                (CompletableFuture<GetTopicsResult>) pendingRequests.remove(requestId);
        if (requestFuture != null) {
            requestFuture.complete(new GetTopicsResult(topics,
                    success.hasTopicsHash() ? success.getTopicsHash() : null,
                    success.isFiltered(),
                    success.isChanged()));
        } else {
            log.warn("{} Received unknown request id from server: {}", ctx.channel(), success.getRequestId());
        }
    }

    @Override
    protected void handleGetSchemaResponse(CommandGetSchemaResponse commandGetSchemaResponse) {
        checkArgument(state == State.Ready);

        long requestId = commandGetSchemaResponse.getRequestId();

        CompletableFuture<CommandGetSchemaResponse> future =
                (CompletableFuture<CommandGetSchemaResponse>) pendingRequests.remove(requestId);
        if (future == null) {
            log.warn("{} Received unknown request id from server: {}", ctx.channel(), requestId);
            return;
        }
        future.complete(new CommandGetSchemaResponse().copyFrom(commandGetSchemaResponse));
    }

    @Override
    protected void handleGetOrCreateSchemaResponse(CommandGetOrCreateSchemaResponse commandGetOrCreateSchemaResponse) {
        checkArgument(state == State.Ready);
        long requestId = commandGetOrCreateSchemaResponse.getRequestId();
        CompletableFuture<CommandGetOrCreateSchemaResponse> future =
                (CompletableFuture<CommandGetOrCreateSchemaResponse>) pendingRequests.remove(requestId);
        if (future == null) {
            log.warn("{} Received unknown request id from server: {}", ctx.channel(), requestId);
            return;
        }
        future.complete(new CommandGetOrCreateSchemaResponse().copyFrom(commandGetOrCreateSchemaResponse));
    }

    Promise<Void> newPromise() {
        return ctx.newPromise();
    }

    public ChannelHandlerContext ctx() {
        return ctx;
    }

    Channel channel() {
        return ctx.channel();
    }

    SocketAddress serverAddrees() {
        return remoteAddress;
    }

    CompletableFuture<Void> connectionFuture() {
        return connectionFuture;
    }

    CompletableFuture<ProducerResponse> sendRequestWithId(ByteBuf cmd, long requestId) {
        return sendRequestAndHandleTimeout(cmd, requestId, RequestType.Command, true);
    }

    private <T> void sendRequestAndHandleTimeout(ByteBuf requestMessage, long requestId,
                                                                 RequestType requestType, boolean flush,
                                                                 TimedCompletableFuture<T> future) {
        pendingRequests.put(requestId, future);
        if (flush) {
            ctx.writeAndFlush(requestMessage).addListener(writeFuture -> {
                if (!writeFuture.isSuccess()) {
                    if (pendingRequests.remove(requestId, future) && !future.isDone()) {
                        log.warn("{} Failed to send {} to broker: {}", ctx.channel(),
                                requestType.getDescription(), writeFuture.cause().getMessage());
                        future.completeExceptionally(writeFuture.cause());
                    }
                }
            });
        } else {
            ctx.write(requestMessage, ctx().voidPromise());
        }
        requestTimeoutQueue.add(new RequestTime(requestId, requestType));
    }

    private <T> CompletableFuture<T> sendRequestAndHandleTimeout(ByteBuf requestMessage, long requestId,
                                                 RequestType requestType, boolean flush) {
        TimedCompletableFuture<T> future = new TimedCompletableFuture<>();
        sendRequestAndHandleTimeout(requestMessage, requestId, requestType, flush, future);
        return future;
    }

    public CompletableFuture<CommandGetLastMessageIdResponse> sendGetLastMessageId(ByteBuf request, long requestId) {
        return sendRequestAndHandleTimeout(request, requestId, RequestType.GetLastMessageId, true);
    }

    public CompletableFuture<Optional<SchemaInfo>> sendGetSchema(ByteBuf request, long requestId) {
        return sendGetRawSchema(request, requestId).thenCompose(commandGetSchemaResponse -> {
            if (commandGetSchemaResponse.hasErrorCode()) {
                // Request has failed
                ServerError rc = commandGetSchemaResponse.getErrorCode();
                if (rc == ServerError.TopicNotFound) {
                    return CompletableFuture.completedFuture(Optional.empty());
                } else {
                    return FutureUtil.failedFuture(
                        getPulsarClientException(rc,
                                buildError(requestId, commandGetSchemaResponse.getErrorMessage())));
                }
            } else {
                return CompletableFuture.completedFuture(
                    Optional.of(SchemaInfoUtil.newSchemaInfo(commandGetSchemaResponse.getSchema())));
            }
        });
    }

    public CompletableFuture<CommandGetSchemaResponse> sendGetRawSchema(ByteBuf request, long requestId) {
        return sendRequestAndHandleTimeout(request, requestId, RequestType.GetSchema, true);
    }

    public CompletableFuture<byte[]> sendGetOrCreateSchema(ByteBuf request, long requestId) {
        CompletableFuture<CommandGetOrCreateSchemaResponse> future = sendRequestAndHandleTimeout(request, requestId,
                RequestType.GetOrCreateSchema, true);
        return future.thenCompose(response -> {
            if (response.hasErrorCode()) {
                // Request has failed
                ServerError rc = response.getErrorCode();
                if (rc == ServerError.TopicNotFound) {
                    return CompletableFuture.completedFuture(SchemaVersion.Empty.bytes());
                } else {
                    return FutureUtil.failedFuture(getPulsarClientException(
                                                           rc, buildError(requestId, response.getErrorMessage())));
                }
            } else {
                return CompletableFuture.completedFuture(response.getSchemaVersion());
            }
        });
    }

    @Override
    protected void handleNewTxnResponse(CommandNewTxnResponse command) {
        TransactionMetaStoreHandler handler = checkAndGetTransactionMetaStoreHandler(command.getTxnidMostBits());
        if (handler != null) {
            handler.handleNewTxnResponse(command);
        }
    }

    @Override
    protected void handleAddPartitionToTxnResponse(CommandAddPartitionToTxnResponse command) {
        TransactionMetaStoreHandler handler = checkAndGetTransactionMetaStoreHandler(command.getTxnidMostBits());
        if (handler != null) {
            handler.handleAddPublishPartitionToTxnResponse(command);
        }
    }

    @Override
    protected void handleAddSubscriptionToTxnResponse(CommandAddSubscriptionToTxnResponse command) {
        TransactionMetaStoreHandler handler = checkAndGetTransactionMetaStoreHandler(command.getTxnidMostBits());
        if (handler != null) {
            handler.handleAddSubscriptionToTxnResponse(command);
        }
    }

    @Override
    protected void handleEndTxnOnPartitionResponse(CommandEndTxnOnPartitionResponse command) {
        TransactionBufferHandler handler = checkAndGetTransactionBufferHandler();
        if (handler != null) {
            handler.handleEndTxnOnTopicResponse(command.getRequestId(), command);
        }
    }

    @Override
    protected void handleEndTxnOnSubscriptionResponse(CommandEndTxnOnSubscriptionResponse command) {
        TransactionBufferHandler handler = checkAndGetTransactionBufferHandler();
        if (handler != null) {
            handler.handleEndTxnOnSubscriptionResponse(command.getRequestId(), command);
        }
    }

    @Override
    protected void handleEndTxnResponse(CommandEndTxnResponse command) {
        TransactionMetaStoreHandler handler = checkAndGetTransactionMetaStoreHandler(command.getTxnidMostBits());
        if (handler != null) {
            handler.handleEndTxnResponse(command);
        }
    }

    @Override
    protected void handleTcClientConnectResponse(CommandTcClientConnectResponse response) {
        checkArgument(state == State.Ready);

        if (log.isDebugEnabled()) {
            log.debug("{} Received tc client connect response "
                    + "from server: {}", ctx.channel(), response.getRequestId());
        }
        long requestId = response.getRequestId();
        CompletableFuture<?> requestFuture = pendingRequests.remove(requestId);

        if (requestFuture != null && !requestFuture.isDone()) {
            if (!response.hasError()) {
                requestFuture.complete(null);
            } else {
                ServerError error = response.getError();
                log.error("Got tc client connect response for request: {}, error: {}, errorMessage: {}",
                        response.getRequestId(), response.getError(), response.getMessage());
                requestFuture.completeExceptionally(getExceptionByServerError(error, response.getMessage()));
            }
        } else {
            log.warn("Tc client connect command has been completed and get response for request: {}",
                    response.getRequestId());
        }
    }

    private TransactionMetaStoreHandler checkAndGetTransactionMetaStoreHandler(long tcId) {
        TransactionMetaStoreHandler handler = transactionMetaStoreHandlers.get(tcId);
        if (handler == null) {
            channel().close();
            log.warn("Close the channel since can't get the transaction meta store handler, will reconnect later.");
        }
        return handler;
    }

    private TransactionBufferHandler checkAndGetTransactionBufferHandler() {
        if (transactionBufferHandler == null) {
            channel().close();
            log.warn("Close the channel since can't get the transaction buffer handler.");
        }
        return transactionBufferHandler;
    }

    public CompletableFuture<CommandWatchTopicListSuccess> newWatchTopicList(
            BaseCommand commandWatchTopicList, long requestId) {
        if (!supportsTopicWatchers) {
            return FutureUtil.failedFuture(
                    new PulsarClientException.NotAllowedException(
                            "Broker does not allow broker side pattern evaluation."));
        }
        return sendRequestAndHandleTimeout(Commands.serializeWithSize(commandWatchTopicList), requestId,
                RequestType.Command, true);
    }

    public CompletableFuture<CommandSuccess> newWatchTopicListClose(
            BaseCommand commandWatchTopicListClose, long requestId) {
        return sendRequestAndHandleTimeout(
                Commands.serializeWithSize(commandWatchTopicListClose), requestId, RequestType.Command, true);
    }

    protected void handleCommandWatchTopicListSuccess(CommandWatchTopicListSuccess commandWatchTopicListSuccess) {
        checkArgument(state == State.Ready);

        if (log.isDebugEnabled()) {
            log.debug("{} Received watchTopicListSuccess response from server: {}",
                    ctx.channel(), commandWatchTopicListSuccess.getRequestId());
        }
        long requestId = commandWatchTopicListSuccess.getRequestId();
        CompletableFuture<CommandWatchTopicListSuccess> requestFuture =
                (CompletableFuture<CommandWatchTopicListSuccess>) pendingRequests.remove(requestId);
        if (requestFuture != null) {
            requestFuture.complete(commandWatchTopicListSuccess);
        } else {
            log.warn("{} Received unknown request id from server: {}",
                    ctx.channel(), commandWatchTopicListSuccess.getRequestId());
        }
    }

    protected void handleCommandWatchTopicUpdate(CommandWatchTopicUpdate commandWatchTopicUpdate) {
        checkArgument(state == State.Ready);

        if (log.isDebugEnabled()) {
            log.debug("{} Received watchTopicUpdate command from server: {}",
                    ctx.channel(), commandWatchTopicUpdate.getWatcherId());
        }
        long watcherId = commandWatchTopicUpdate.getWatcherId();
        TopicListWatcher watcher = topicListWatchers.get(watcherId);
        if (watcher != null) {
            watcher.handleCommandWatchTopicUpdate(commandWatchTopicUpdate);
        } else {
            log.warn("{} Received topic list update for unknown watcher from server: {}", ctx.channel(), watcherId);
        }
    }

    /**
     * check serverError and take appropriate action.
     * <ul>
     * <li>InternalServerError: close connection immediately</li>
     * <li>TooManyRequest: received error count is more than maxNumberOfRejectedRequestPerConnection in
     * #rejectedRequestResetTimeSec</li>
     * </ul>
     *
     * @param error
     * @param errMsg
     */
    private void checkServerError(ServerError error, String errMsg) {
        if (ServerError.ServiceNotReady.equals(error)) {
            log.error("{} Close connection because received internal-server error {}", ctx.channel(), errMsg);
            ctx.close();
        } else if (ServerError.TooManyRequests.equals(error)) {
            incrementRejectsAndMaybeClose();
        }
    }

    private void incrementRejectsAndMaybeClose() {
        long rejectedRequests = NUMBER_OF_REJECTED_REQUESTS_UPDATER.getAndIncrement(this);
        if (rejectedRequests == 0) {
            // schedule timer
            eventLoopGroup.schedule(() -> NUMBER_OF_REJECTED_REQUESTS_UPDATER.set(ClientCnx.this, 0),
                                    rejectedRequestResetTimeSec, TimeUnit.SECONDS);
        } else if (rejectedRequests >= maxNumberOfRejectedRequestPerConnection) {
            log.error("{} Close connection because received {} rejected request in {} seconds ", ctx.channel(),
                      NUMBER_OF_REJECTED_REQUESTS_UPDATER.get(ClientCnx.this), rejectedRequestResetTimeSec);
            ctx.close();
        }
    }

    void registerConsumer(final long consumerId, final ConsumerImpl<?> consumer) {
        consumers.put(consumerId, consumer);
    }

    void registerProducer(final long producerId, final ProducerImpl<?> producer) {
        producers.put(producerId, producer);
    }

    void registerTransactionMetaStoreHandler(final long transactionMetaStoreId,
                                             final TransactionMetaStoreHandler handler) {
        transactionMetaStoreHandlers.put(transactionMetaStoreId, handler);
    }

    void registerTopicListWatcher(final long watcherId, final TopicListWatcher watcher) {
        topicListWatchers.put(watcherId, watcher);

    }

    public void registerTransactionBufferHandler(final TransactionBufferHandler handler) {
        transactionBufferHandler = handler;
    }

    void removeProducer(final long producerId) {
        producers.remove(producerId);
    }

    void removeConsumer(final long consumerId) {
        consumers.remove(consumerId);
    }

    void removeTopicListWatcher(final long watcherId) {
        topicListWatchers.remove(watcherId);
    }

    void setTargetBroker(InetSocketAddress targetBrokerAddress) {
        this.proxyToTargetBrokerAddress = String.format("%s:%d", targetBrokerAddress.getHostString(),
                targetBrokerAddress.getPort());
    }

     void setRemoteHostName(String remoteHostName) {
        this.remoteHostName = remoteHostName;
    }

    private String buildError(long requestId, String errorMsg) {
        return new StringBuilder().append("{\"errorMsg\":\"").append(errorMsg)
            .append("\",\"reqId\":").append(requestId)
            .append(", \"remote\":\"").append(remoteAddress)
            .append("\", \"local\":\"").append(localAddress)
            .append("\"}").toString();
    }

    public static PulsarClientException getPulsarClientException(ServerError error, String errorMsg) {
        switch (error) {
        case AuthenticationError:
            return new PulsarClientException.AuthenticationException(errorMsg);
        case AuthorizationError:
            return new PulsarClientException.AuthorizationException(errorMsg);
        case ProducerBusy:
            return new PulsarClientException.ProducerBusyException(errorMsg);
        case ConsumerBusy:
            return new PulsarClientException.ConsumerBusyException(errorMsg);
        case MetadataError:
            return new PulsarClientException.BrokerMetadataException(errorMsg);
        case PersistenceError:
            return new PulsarClientException.BrokerPersistenceException(errorMsg);
        case ServiceNotReady:
            return new PulsarClientException.LookupException(errorMsg);
        case TooManyRequests:
            return new PulsarClientException.TooManyRequestsException(errorMsg);
        case ProducerBlockedQuotaExceededError:
            return new PulsarClientException.ProducerBlockedQuotaExceededError(errorMsg);
        case ProducerBlockedQuotaExceededException:
            return new PulsarClientException.ProducerBlockedQuotaExceededException(errorMsg);
        case TopicTerminatedError:
            return new PulsarClientException.TopicTerminatedException(errorMsg);
        case IncompatibleSchema:
            return new PulsarClientException.IncompatibleSchemaException(errorMsg);
        case TopicNotFound:
            return new PulsarClientException.TopicDoesNotExistException(errorMsg);
        case ConsumerAssignError:
            return new PulsarClientException.ConsumerAssignException(errorMsg);
        case NotAllowedError:
            return new PulsarClientException.NotAllowedException(errorMsg);
        case TransactionConflict:
            return new PulsarClientException.TransactionConflictException(errorMsg);
        case ProducerFenced:
            return new PulsarClientException.ProducerFencedException(errorMsg);
        case UnknownError:
        default:
            return new PulsarClientException(errorMsg);
        }
    }

    public void close() {
       if (ctx != null) {
           ctx.close();
       }
    }

    protected void closeWithException(Throwable e) {
       if (ctx != null) {
           connectionFuture.completeExceptionally(e);
           ctx.close();
       }
    }

    private void checkRequestTimeout() {
        while (!requestTimeoutQueue.isEmpty()) {
            RequestTime request = requestTimeoutQueue.peek();
            if (request == null || !request.isTimedOut(operationTimeoutMs)) {
                // if there is no request that is timed out then exit the loop
                break;
            }
            request = requestTimeoutQueue.poll();
            TimedCompletableFuture<?> requestFuture = pendingRequests.get(request.requestId);
            if (requestFuture != null
                    && !requestFuture.hasGotResponse()) {
                pendingRequests.remove(request.requestId, requestFuture);
                if (!requestFuture.isDone()) {
                    String timeoutMessage = String.format(
                            "%s timeout {'durationMs': '%d', 'reqId':'%d', 'remote':'%s', 'local':'%s'}",
                            request.requestType.getDescription(), operationTimeoutMs,
                            request.requestId, remoteAddress, localAddress);
                    if (requestFuture.completeExceptionally(new TimeoutException(timeoutMessage))) {
                        if (request.requestType == RequestType.Lookup) {
                            incrementRejectsAndMaybeClose();
                        }
                        log.warn("{} {}", ctx.channel(), timeoutMessage);
                    }
                }
            }
        }
    }

    private static final Logger log = LoggerFactory.getLogger(ClientCnx.class);

    /**
     * Check client connection is now free. This method will not change the state to idle.
     * @return true if the connection is eligible.
     */
    public boolean idleCheck() {
        if (pendingRequests != null && !pendingRequests.isEmpty()) {
            return false;
        }
        if (waitingLookupRequests != null  && !waitingLookupRequests.isEmpty()) {
            return false;
        }
        if (!consumers.isEmpty()) {
            return false;
        }
        if (!producers.isEmpty()) {
            return false;
        }
        if (!transactionMetaStoreHandlers.isEmpty()) {
            return false;
        }
        return true;
    }
}

redpajama_set_name: RedPajamaGithub


text:
package com.jayanslow.projection.world.models;

import javax.vecmath.Matrix3f;
import javax.vecmath.Tuple3d;
import javax.vecmath.Tuple3f;
import javax.vecmath.Vector3f;

public class Rotation3f extends Tuple3f {
	private static final long	serialVersionUID	= 7372169161188811962L;

	public Rotation3f() {
		super();
	}

	public Rotation3f(float x, float y, float z) {
		super(x, y, z);
	}

	public Rotation3f(float[] t) {
		super(t);
	}

	public Rotation3f(Tuple3d t1) {
		super(t1);
	}

	public Rotation3f(Tuple3f t1) {
		super(t1);
	}

	public Matrix3f getRotationMatrix() {
		Matrix3f m = new Matrix3f(), m2 = new Matrix3f();

		// m = Rz(angleZ)
		m.rotZ(z);

		// m = Rz(angleZ) . Ry(angleX)
		m2.rotX(x);
		m.mul(m2);

		// m = Rz(angleZ) . Ry(angleX) . Rx(angleY)
		m2.rotY(y);
		m.mul(m2);

		return m;
	}

	public void inverseRotate(Vector3f v) {
		Matrix3f m = getRotationMatrix();
		m.invert();
		m.transform(v);
	}

	public void rotate(Vector3f v) {
		Matrix3f m = getRotationMatrix();
		m.transform(v);
	}

	public Rotation3f toDegrees() {
		return new Rotation3f((float) Math.toDegrees(x), (float) Math.toDegrees(y), (float) Math.toDegrees(z));
	}

	public Rotation3f toRadians() {
		return new Rotation3f((float) Math.toRadians(x), (float) Math.toRadians(y), (float) Math.toRadians(z));
	}
}


text:
<!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN" "http://www.w3.org/TR/html4/loose.dtd">

    <title>H-Index</title>
    <meta name="date" content="2015-10-15">
    <link rel="stylesheet" type="text/css" href="../stylesheet.css" title="Style">

JavaScript is disabled on your browser.

Skip navigation links

Overview

Package

Class

Tree

Deprecated

Index

Help

Prev Letter

Next Letter

Frames

No Frames

All Classes

A C D F G H I P R S T

H

HandlerMissingHandler - Class in de.deadmist.deadSimpleHttp.handlers

This handler is called when no handler matches the URL

                It sends an 501 Not Implemented error message stating the URL and that no handler has been configured.

HandlerMissingHandler(String)
            - Constructor for class de.deadmist.deadSimpleHttp.handlers.HandlerMissingHandler

Creates new handler

HTTPServer -
            Class in de.deadmist.deadSimpleHttp.server

A simple HTTP server, functionality is provided through RequestHandlers that
                are matched to the requested URL.

HTTPServer(int,
            int, HashMap<String, RequestHandler>) - Constructor for class
            de.deadmist.deadSimpleHttp.server.HTTPServer

Creates a new server instance.

HTTPServer(int, int)
            - Constructor for class de.deadmist.deadSimpleHttp.server.HTTPServer

Creates a new server instance.

HTTPSServer -
            Class in de.deadmist.deadSimpleHttp.server

A simple HTTPS server, functionality is provided through RequestHandlers that
                are matched to the requested URL.

HTTPSServer(int,
            int, HashMap<String, RequestHandler>, String, String) - Constructor for class
            de.deadmist.deadSimpleHttp.server.HTTPSServer

Creates a new server instance.

HTTPSServer(int,
            int, String, String) - Constructor for class de.deadmist.deadSimpleHttp.server.HTTPSServer

Creates a new server instance.

    A C D F G H I P R S T

Skip navigation links

Overview

Package

Class

Tree

Deprecated

Index

Help

Prev Letter

Next Letter

Frames

No Frames

All Classes


text:
<!DOCTYPE html>

<meta charset="utf-8">
<meta name="description" content="listings">
<meta name="viewport" content="width=device-width">
<title>listings | Warren Liu</title>
<link href="../assets/css/rst.css" rel="stylesheet" type="text/css">
<link href="../assets/css/poole.css" rel="stylesheet" type="text/css">
<link href="../assets/css/lanyon.css" rel="stylesheet" type="text/css">
<link href="../assets/css/code.css" rel="stylesheet" type="text/css">
<link href="../assets/css/custom.css" rel="stylesheet" type="text/css">
<link rel="stylesheet" href="//fonts.googleapis.com/css?family=PT+Serif:400,400italic,700%7CPT+Sans:400">
<link rel="alternate" type="application/rss+xml" title="RSS" href="../rss.xml">
<link rel="canonical" href="https://wudong.graceliu.uk/listings/">

    Skip to main content

    <input type="checkbox" class="sidebar-checkbox" id="sidebar-checkbox">

A reserved Nikola theme that places the utmost gravity on content with a hidden drawer. Made by @mdo for Jekyll,
            ported to Nikola by @ralsina.

    Archive
        Tags
        RSS feed

      Warren Liu

listings

Contents © 2017         Warren Liu - Powered by         Nikola


text:
package eu.drus.jpa.unit.spi;

import java.lang.reflect.Method;
import java.util.Optional;

public interface TestInvocation {

    Class<?> getTestClass();

    ExecutionContext getContext();

    Optional<Method> getTestMethod();

    Optional<Object> getTestInstance();

    Optional<Throwable> getException();

    FeatureResolver getFeatureResolver();
}


text:
package br.com.lience.service;

import br.com.lience.model.Usuario;
import org.hibernate.Session;

import javax.ejb.Stateless;
import javax.inject.Inject;
import javax.persistence.EntityManager;
import javax.persistence.Persistence;

/**
 * A simple CDI service which is able to say hello to someone
 *
 * @author Marco Aurelyo
 */
@Stateless
public class UsuarioService {

    @Inject
    private EntityManager em;

    public String listar() {
        Usuario usuario = new Usuario();
        usuario.setNoUsuario("nome_usuario_cadastrado");
        usuario.setNoSenha("senhaMd5");

        Session session = (Session) em.getDelegate();
        session.persist(usuario);
        return "Listar Usuários Package Service :)" + usuario.getNoUsuario();
    }
}


text:
package parser

import "github.com/cirbo-lang/cirbo/source"

type tokenPeeker struct {
	Iter   *tokenIterator
	Peeked Token
}

func (p *tokenPeeker) Peek() Token {
	for p.Peeked.Type == TokenNil || p.Peeked.Type == TokenWhitespace || p.Peeked.Type == TokenComment {
		p.Peeked = p.Iter.Next()
	}
	return p.Peeked
}

func (p *tokenPeeker) Read() Token {
	ret := p.Peek()
	p.Peeked.Type = TokenNil
	return ret
}

// PeekIdent checks if the next token is an unquoted identifier, and if so it
// returns the identifier's name as a string. If the next token is not an
// identifier, or if it is quoted with backticks, it returns an empty string.
func (p *tokenPeeker) PeekKeyword() string {
	next := p.Peek()
	if next.Type != TokenIdent {
		return ""
	}
	got := string(next.Bytes)
	if got[0] == '`' {
		// this is a `quoted` ident, which means it can never be a keyword
		// (user can write e.g. `import` to escape the special meaning of
		// the import keyword.)
		return ""
	}
	return got
}

func (p *tokenPeeker) PeekRange() source.Range {
	return p.Peek().Range
}

func (p *tokenPeeker) EOF() bool {
	return p.Peek().Type == TokenEOF
}

type tokenIterator struct {
	Tokens Tokens
	Pos    int
}

func newTokenIterator(tokens Tokens) *tokenIterator {
	return &tokenIterator{
		Tokens: tokens,
		Pos:    0,
	}
}

func (i *tokenIterator) Next() Token {
	ret := i.Tokens[i.Pos]
	if i.Pos < (len(i.Tokens) - 1) {
		// When we reach the end we will just keep returning the final token
		// forever, since we assume it will be a TokenEOF.
		i.Pos++
	}
	return ret
}


text:
#pragma once

//! @file common.h Global constants for liblsl

#if defined(LIBLSL_FFI)
// Skip any typedefs that might confuse a FFI header parser, e.g. cffi
#elif defined(_MSC_VER) && _MSC_VER < 1600
typedef signed char int8_t;
typedef signed short int16_t;
typedef signed int int32_t;
typedef signed long long int64_t;
typedef unsigned int uint32_t;
#else
#include <stdint.h>
#endif

#if defined(_MSC_VER) && _MSC_VER < 1900
#define __func__ __FUNCTION__
#endif

/// LIBLSL_C_API expands function attributes needed for the linker
#if defined(LIBLSL_STATIC) || defined(LIBLSL_FFI)
#define LIBLSL_C_API
#elif defined _WIN32 || defined __CYGWIN__
#if defined LIBLSL_EXPORTS
#define LIBLSL_C_API __declspec(dllexport)
#else
#define LIBLSL_C_API __declspec(dllimport)
#ifndef LSLNOAUTOLINK
#pragma comment(lib, "lsl.lib")
#endif
#endif
#pragma warning(disable : 4275)
#else // Linux / OS X
#define LIBLSL_C_API __attribute__((visibility("default")))
#endif

/// Constant to indicate that a stream has variable sampling rate.
#define LSL_IRREGULAR_RATE 0.0

/**
 * Constant to indicate that a sample has the next successive time stamp.
 *
 * This is an optional optimization to transmit less data per sample.
 * The stamp is then deduced from the preceding one according to the stream's
 * sampling rate (in the case of an irregular rate, the same time stamp as
 * before will is assumed). */
#define LSL_DEDUCED_TIMESTAMP -1.0

/// A very large time value (ca. 1 year); can be used in timeouts.
#define LSL_FOREVER 32000000.0

/**
 * Constant to indicate that there is no preference about how a data stream
 * shall be chunked for transmission.
 * (can be used for the chunking parameters in the inlet or the outlet).
 */
#define LSL_NO_PREFERENCE 0

/// Data format of a channel (each transmitted sample holds an array of channels), 4 bytes wide
typedef enum {
	/** For up to 24-bit precision measurements in the appropriate physical unit (e.g., microvolts).
	 * Integers from -16777216 to 16777216 are represented accurately. */
	cft_float32 = 1,
	/** For universal numeric data as long as permitted by network & disk budget.
	 * The largest representable integer is 53-bit. */
	cft_double64 = 2,
	/** For variable-length ASCII strings or data blobs, such as video frames, complex event
	   descriptions, etc. */
	cft_string = 3,
	/** For high-rate digitized formats that require 32-bit precision.
	 * Depends critically on meta-data to represent meaningful units.
	 * Useful for application event codes or other coded data. */
	cft_int32 = 4,
	/** For very high rate signals (40Khz+) or consumer-grade audio.
	 * For professional audio float is recommended. */
	cft_int16 = 5,
	/// For binary signals or other coded data. Not recommended for encoding string data.
	cft_int8 = 6,
	/** 64 bit integers. Support for this type is not yet exposed in all languages.
	 * Also, some builds of liblsl will not be able to send or receive data of this type. */
	cft_int64 = 7,
	/// Can not be transmitted.
	cft_undefined = 0,

	// prevent compilers from assuming an instance fits in a single byte
	_cft_maxval = 0x7f000000
} lsl_channel_format_t;

// Abort compilation if lsl_channel_format_t isn't exactly 4 bytes wide
#if defined(__STDC_VERSION__) && __STDC_VERSION__ >= 201112L
_Static_assert(sizeof(lsl_channel_format_t) == 4, "lsl_channel_format_t size breaks the LSL ABI");
#elif defined(__cplusplus) && __cplusplus >= 201103L
static_assert (sizeof(lsl_channel_format_t) == 4, "lsl_channel_format_t size breaks the LSL ABI");
#elif !defined(LIBLSL_FFI)
static char _lsl_channel_format_size_check[1 - 2*!(sizeof(lsl_channel_format_t)==4)];
#endif

/// Post-processing options for stream inlets.
typedef enum {
	/** No automatic post-processing; return the ground-truth time stamps for manual
	   post-processing. This is the default behavior of the inlet. */
	proc_none = 0,

	/** Perform automatic clock synchronization; equivalent to manually adding the time_correction()
	   value to the received time stamps. */
	proc_clocksync = 1,

	/** Remove jitter from time stamps.
	 *
	 * This will apply a smoothing algorithm to the received time stamps; the smoothing needs to see
	 * a minimum number of samples (30-120 seconds worst-case) until the remaining jitter is
	 * consistently below 1ms. */
	proc_dejitter = 2,

	/** Force the time-stamps to be monotonically ascending.
	 *
	 * Only makes sense if timestamps are dejittered. */
	proc_monotonize = 4,

	/** Post-processing is thread-safe (same inlet can be read from by multiple threads);
	 * uses somewhat more CPU. */
	proc_threadsafe = 8,

	/// The combination of all possible post-processing options.
	proc_ALL = 1 | 2 | 4 | 8,

	// prevent compilers from assuming an instance fits in a single byte
	_proc_maxval = 0x7f000000
} lsl_processing_options_t;

/// Possible error codes.
typedef enum {
	/// No error occurred
	lsl_no_error = 0,

	/// The operation failed due to a timeout.
	lsl_timeout_error = -1,

	/// The stream has been lost.
	lsl_lost_error = -2,

	/// An argument was incorrectly specified (e.g., wrong format or wrong length).
	lsl_argument_error = -3,

	/// Some other internal error has happened.
	lsl_internal_error = -4,

	// prevent compilers from assuming an instance fits in a single byte
	_lsl_error_code_maxval = 0x7f000000
} lsl_error_code_t;

/// Return an explanation for the last error
extern LIBLSL_C_API const char *lsl_last_error(void);

/**
 * LSL version the binary was compiled against
 *
 * Used either to check if the same version is used
 * (`if(lsl_protocol_version()!=LIBLSL_COMPILE_HEADER_VERSION`) …
 * or to require a certain set of features:
 * ```
 * #if LIBLSL_COMPILE_HEADER_VERSION > 113
 * do_stuff();
 * #endif
 * ```
 * */
#define LIBLSL_COMPILE_HEADER_VERSION 114

/**
 * Protocol version.
 *
 * The major version is `protocol_version() / 100;`
 * The minor version is `protocol_version() % 100;`
 *
 * Clients with different minor versions are protocol-compatible while clients
 * with different major versions will refuse to work together.
 */
extern LIBLSL_C_API int32_t lsl_protocol_version();

/**
 * Version of the liblsl library.
 *
 * The major version is `library_version() / 100;`
 * The minor version is `library_version() % 100;`
 */
extern LIBLSL_C_API int32_t lsl_library_version();

/**
 * Get a string containing library information.
 *
 * The format of the string shouldn't be used for anything important except giving a debugging
 * person a good idea which exact library version is used. */
extern LIBLSL_C_API const char *lsl_library_info(void);

/**
 * Obtain a local system time stamp in seconds.
 *
 * The resolution is better than a millisecond.
 * This reading can be used to assign time stamps to samples as they are being acquired.
 * If the "age" of a sample is known at a particular time (e.g., from USB transmission
 * delays), it can be used as an offset to lsl_local_clock() to obtain a better estimate of
 * when a sample was actually captured. See lsl_push_sample() for a use case.
 */
extern LIBLSL_C_API double lsl_local_clock();

/**
 * Deallocate a string that has been transferred to the application.
 *
 * Rarely used: the only use case is to deallocate the contents of
 * string-valued samples received from LSL in an application where
 * no free() method is available (e.g., in some scripting languages).
 */
extern LIBLSL_C_API void lsl_destroy_string(char *s);