text:
package org.apache.kafka.clients;

import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.AuthenticationException;
import org.apache.kafka.common.errors.DisconnectException;
import org.apache.kafka.common.errors.UnsupportedVersionException;
import org.apache.kafka.common.message.ApiVersionsResponseData.ApiVersionsResponseKey;
import org.apache.kafka.common.metrics.Sensor;
import org.apache.kafka.common.network.ChannelState;
import org.apache.kafka.common.network.NetworkReceive;
import org.apache.kafka.common.network.Selectable;
import org.apache.kafka.common.network.Send;
import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.protocol.CommonFields;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.protocol.types.SchemaException;
import org.apache.kafka.common.protocol.types.Struct;
import org.apache.kafka.common.requests.AbstractRequest;
import org.apache.kafka.common.requests.AbstractResponse;
import org.apache.kafka.common.requests.ApiVersionsRequest;
import org.apache.kafka.common.requests.ApiVersionsResponse;
import org.apache.kafka.common.requests.MetadataRequest;
import org.apache.kafka.common.requests.MetadataResponse;
import org.apache.kafka.common.requests.RequestHeader;
import org.apache.kafka.common.requests.ResponseHeader;
import org.apache.kafka.common.security.authenticator.SaslClientAuthenticator;
import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.common.utils.Utils;
import org.slf4j.Logger;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.nio.BufferUnderflowException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Random;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;

/**
 * A network client for asynchronous request/response network i/o. This is an internal class used to implement the
 * user-facing producer and consumer clients.
 * <p>
 * This class is not thread-safe!
 */
public class NetworkClient implements KafkaClient {

    private enum State {
        ACTIVE,
        CLOSING,
        CLOSED
    }

    private final Logger log;

    /* the selector used to perform network i/o */
    private final Selectable selector;

    private final MetadataUpdater metadataUpdater;

    private final Random randOffset;

    /* the state of each node's connection */
    private final ClusterConnectionStates connectionStates;

    /* the set of requests currently being sent or awaiting a response */
    private final InFlightRequests inFlightRequests;

    /* the socket send buffer size in bytes */
    private final int socketSendBuffer;

    /* the socket receive size buffer in bytes */
    private final int socketReceiveBuffer;

    /* the client id used to identify this client in requests to the server */
    private final String clientId;

    /* the current correlation id to use when sending requests to servers */
    private int correlation;

    /* default timeout for individual requests to await acknowledgement from servers */
    private final int defaultRequestTimeoutMs;

    /* time in ms to wait before retrying to create connection to a server */
    private final long reconnectBackoffMs;

    private final ClientDnsLookup clientDnsLookup;

    private final Time time;

    /**
     * True if we should send an ApiVersionRequest when first connecting to a broker.
     */
    private final boolean discoverBrokerVersions;

    private final ApiVersions apiVersions;

    private final Map<String, ApiVersionsRequest.Builder> nodesNeedingApiVersionsFetch = new HashMap<>();

    private final List<ClientResponse> abortedSends = new LinkedList<>();

    private final Sensor throttleTimeSensor;

    private final AtomicReference<State> state;

    public NetworkClient(Selectable selector,
                         Metadata metadata,
                         String clientId,
                         int maxInFlightRequestsPerConnection,
                         long reconnectBackoffMs,
                         long reconnectBackoffMax,
                         int socketSendBuffer,
                         int socketReceiveBuffer,
                         int defaultRequestTimeoutMs,
                         ClientDnsLookup clientDnsLookup,
                         Time time,
                         boolean discoverBrokerVersions,
                         ApiVersions apiVersions,
                         LogContext logContext) {
        this(null,
             metadata,
             selector,
             clientId,
             maxInFlightRequestsPerConnection,
             reconnectBackoffMs,
             reconnectBackoffMax,
             socketSendBuffer,
             socketReceiveBuffer,
             defaultRequestTimeoutMs,
             clientDnsLookup,
             time,
             discoverBrokerVersions,
             apiVersions,
             null,
             logContext);
    }

    public NetworkClient(Selectable selector,
            Metadata metadata,
            String clientId,
            int maxInFlightRequestsPerConnection,
            long reconnectBackoffMs,
            long reconnectBackoffMax,
            int socketSendBuffer,
            int socketReceiveBuffer,
            int defaultRequestTimeoutMs,
            ClientDnsLookup clientDnsLookup,
            Time time,
            boolean discoverBrokerVersions,
            ApiVersions apiVersions,
            Sensor throttleTimeSensor,
            LogContext logContext) {
        this(null,
             metadata,
             selector,
             clientId,
             maxInFlightRequestsPerConnection,
             reconnectBackoffMs,
             reconnectBackoffMax,
             socketSendBuffer,
             socketReceiveBuffer,
             defaultRequestTimeoutMs,
             clientDnsLookup,
             time,
             discoverBrokerVersions,
             apiVersions,
             throttleTimeSensor,
             logContext);
    }

    public NetworkClient(Selectable selector,
                         MetadataUpdater metadataUpdater,
                         String clientId,
                         int maxInFlightRequestsPerConnection,
                         long reconnectBackoffMs,
                         long reconnectBackoffMax,
                         int socketSendBuffer,
                         int socketReceiveBuffer,
                         int defaultRequestTimeoutMs,
                         ClientDnsLookup clientDnsLookup,
                         Time time,
                         boolean discoverBrokerVersions,
                         ApiVersions apiVersions,
                         LogContext logContext) {
        this(metadataUpdater,
             null,
             selector,
             clientId,
             maxInFlightRequestsPerConnection,
             reconnectBackoffMs,
             reconnectBackoffMax,
             socketSendBuffer,
             socketReceiveBuffer,
             defaultRequestTimeoutMs,
             clientDnsLookup,
             time,
             discoverBrokerVersions,
             apiVersions,
             null,
             logContext);
    }

    private NetworkClient(MetadataUpdater metadataUpdater,
                          Metadata metadata,
                          Selectable selector,
                          String clientId,
                          int maxInFlightRequestsPerConnection,
                          long reconnectBackoffMs,
                          long reconnectBackoffMax,
                          int socketSendBuffer,
                          int socketReceiveBuffer,
                          int defaultRequestTimeoutMs,
                          ClientDnsLookup clientDnsLookup,
                          Time time,
                          boolean discoverBrokerVersions,
                          ApiVersions apiVersions,
                          Sensor throttleTimeSensor,
                          LogContext logContext) {
        /* It would be better if we could pass `DefaultMetadataUpdater` from the public constructor, but it's not
         * possible because `DefaultMetadataUpdater` is an inner class and it can only be instantiated after the
         * super constructor is invoked.
         */
        if (metadataUpdater == null) {
            if (metadata == null)
                throw new IllegalArgumentException("`metadata` must not be null");
            this.metadataUpdater = new DefaultMetadataUpdater(metadata);
        } else {
            this.metadataUpdater = metadataUpdater;
        }
        this.selector = selector;
        this.clientId = clientId;
        this.inFlightRequests = new InFlightRequests(maxInFlightRequestsPerConnection);
        this.connectionStates = new ClusterConnectionStates(reconnectBackoffMs, reconnectBackoffMax, logContext);
        this.socketSendBuffer = socketSendBuffer;
        this.socketReceiveBuffer = socketReceiveBuffer;
        this.correlation = 0;
        this.randOffset = new Random();
        this.defaultRequestTimeoutMs = defaultRequestTimeoutMs;
        this.reconnectBackoffMs = reconnectBackoffMs;
        this.time = time;
        this.discoverBrokerVersions = discoverBrokerVersions;
        this.apiVersions = apiVersions;
        this.throttleTimeSensor = throttleTimeSensor;
        this.log = logContext.logger(NetworkClient.class);
        this.clientDnsLookup = clientDnsLookup;
        this.state = new AtomicReference<>(State.ACTIVE);
    }

    /**
     * Begin connecting to the given node, return true if we are already connected and ready to send to that node.
     *
     * @param node The node to check
     * @param now The current timestamp
     * @return True if we are ready to send to the given node
     */
    @Override
    public boolean ready(Node node, long now) {
        if (node.isEmpty())
            throw new IllegalArgumentException("Cannot connect to empty node " + node);

        if (isReady(node, now))
            return true;

        if (connectionStates.canConnect(node.idString(), now))
            // if we are interested in sending to a node and we don't have a connection to it, initiate one
            initiateConnect(node, now);

        return false;
    }

    // Visible for testing
    boolean canConnect(Node node, long now) {
        return connectionStates.canConnect(node.idString(), now);
    }

    /**
     * Disconnects the connection to a particular node, if there is one.
     * Any pending ClientRequests for this connection will receive disconnections.
     *
     * @param nodeId The id of the node
     */
    @Override
    public void disconnect(String nodeId) {
        if (connectionStates.isDisconnected(nodeId))
            return;

        selector.close(nodeId);
        long now = time.milliseconds();

        cancelInFlightRequests(nodeId, now, abortedSends);

        connectionStates.disconnected(nodeId, now);

        if (log.isTraceEnabled()) {
            log.trace("Manually disconnected from {}. Aborted in-flight requests: {}.", nodeId, inFlightRequests);
        }
    }

    private void cancelInFlightRequests(String nodeId, long now, Collection<ClientResponse> responses) {
        Iterable<InFlightRequest> inFlightRequests = this.inFlightRequests.clearAll(nodeId);
        for (InFlightRequest request : inFlightRequests) {
            log.trace("Cancelled request {} {} with correlation id {} due to node {} being disconnected",
                    request.header.apiKey(), request.request, request.header.correlationId(), nodeId);

            if (!request.isInternalRequest) {
                if (responses != null)
                    responses.add(request.disconnected(now, null));
            } else if (request.header.apiKey() == ApiKeys.METADATA) {
                metadataUpdater.handleFailedRequest(now, Optional.empty());
            }
        }
    }

    /**
     * Closes the connection to a particular node (if there is one).
     * All requests on the connection will be cleared.  ClientRequest callbacks will not be invoked
     * for the cleared requests, nor will they be returned from poll().
     *
     * @param nodeId The id of the node
     */
    @Override
    public void close(String nodeId) {
        selector.close(nodeId);
        long now = time.milliseconds();
        cancelInFlightRequests(nodeId, now, null);
        connectionStates.remove(nodeId);
    }

    /**
     * Returns the number of milliseconds to wait, based on the connection state, before attempting to send data. When
     * disconnected, this respects the reconnect backoff time. When connecting or connected, this handles slow/stalled
     * connections.
     *
     * @param node The node to check
     * @param now The current timestamp
     * @return The number of milliseconds to wait.
     */
    @Override
    public long connectionDelay(Node node, long now) {
        return connectionStates.connectionDelay(node.idString(), now);
    }

    // Return the remaining throttling delay in milliseconds if throttling is in progress. Return 0, otherwise.
    // This is for testing.
    public long throttleDelayMs(Node node, long now) {
        return connectionStates.throttleDelayMs(node.idString(), now);
    }

    /**
     * Return the poll delay in milliseconds based on both connection and throttle delay.
     * @param node the connection to check
     * @param now the current time in ms
     */
    @Override
    public long pollDelayMs(Node node, long now) {
        return connectionStates.pollDelayMs(node.idString(), now);
    }

    /**
     * Check if the connection of the node has failed, based on the connection state. Such connection failure are
     * usually transient and can be resumed in the next {@link #ready(org.apache.kafka.common.Node, long)} }
     * call, but there are cases where transient failures needs to be caught and re-acted upon.
     *
     * @param node the node to check
     * @return true iff the connection has failed and the node is disconnected
     */
    @Override
    public boolean connectionFailed(Node node) {
        return connectionStates.isDisconnected(node.idString());
    }

    /**
     * Check if authentication to this node has failed, based on the connection state. Authentication failures are
     * propagated without any retries.
     *
     * @param node the node to check
     * @return an AuthenticationException iff authentication has failed, null otherwise
     */
    @Override
    public AuthenticationException authenticationException(Node node) {
        return connectionStates.authenticationException(node.idString());
    }

    /**
     * Check if the node with the given id is ready to send more requests.
     *
     * @param node The node
     * @param now The current time in ms
     * @return true if the node is ready
     */
    @Override
    public boolean isReady(Node node, long now) {
        // if we need to update our metadata now declare all requests unready to make metadata requests first
        // priority
        return !metadataUpdater.isUpdateDue(now) && canSendRequest(node.idString(), now);
    }

    /**
     * Are we connected and ready and able to send more requests to the given connection?
     *
     * @param node The node
     * @param now the current timestamp
     */
    private boolean canSendRequest(String node, long now) {
        return connectionStates.isReady(node, now) && selector.isChannelReady(node) &&
            inFlightRequests.canSendMore(node);
    }

    /**
     * Queue up the given request for sending. Requests can only be sent out to ready nodes.
     * @param request The request
     * @param now The current timestamp
     */
    @Override
    public void send(ClientRequest request, long now) {
        doSend(request, false, now);
    }

    // package-private for testing
    void sendInternalMetadataRequest(MetadataRequest.Builder builder, String nodeConnectionId, long now) {
        ClientRequest clientRequest = newClientRequest(nodeConnectionId, builder, now, true);
        doSend(clientRequest, true, now);
    }

    private void doSend(ClientRequest clientRequest, boolean isInternalRequest, long now) {
        ensureActive();
        String nodeId = clientRequest.destination();
        if (!isInternalRequest) {
            // If this request came from outside the NetworkClient, validate
            // that we can send data.  If the request is internal, we trust
            // that internal code has done this validation.  Validation
            // will be slightly different for some internal requests (for
            // example, ApiVersionsRequests can be sent prior to being in
            // READY state.)
            if (!canSendRequest(nodeId, now))
                throw new IllegalStateException("Attempt to send a request to node " + nodeId + " which is not ready.");
        }
        AbstractRequest.Builder<?> builder = clientRequest.requestBuilder();
        try {
            NodeApiVersions versionInfo = apiVersions.get(nodeId);
            short version;
            // Note: if versionInfo is null, we have no server version information. This would be
            // the case when sending the initial ApiVersionRequest which fetches the version
            // information itself.  It is also the case when discoverBrokerVersions is set to false.
            if (versionInfo == null) {
                version = builder.latestAllowedVersion();
                if (discoverBrokerVersions && log.isTraceEnabled())
                    log.trace("No version information found when sending {} with correlation id {} to node {}. " +
                            "Assuming version {}.", clientRequest.apiKey(), clientRequest.correlationId(), nodeId, version);
            } else {
                version = versionInfo.latestUsableVersion(clientRequest.apiKey(), builder.oldestAllowedVersion(),
                        builder.latestAllowedVersion());
            }
            // The call to build may also throw UnsupportedVersionException, if there are essential
            // fields that cannot be represented in the chosen version.
            doSend(clientRequest, isInternalRequest, now, builder.build(version));
        } catch (UnsupportedVersionException unsupportedVersionException) {
            // If the version is not supported, skip sending the request over the wire.
            // Instead, simply add it to the local queue of aborted requests.
            log.debug("Version mismatch when attempting to send {} with correlation id {} to {}", builder,
                    clientRequest.correlationId(), clientRequest.destination(), unsupportedVersionException);
            ClientResponse clientResponse = new ClientResponse(clientRequest.makeHeader(builder.latestAllowedVersion()),
                    clientRequest.callback(), clientRequest.destination(), now, now,
                    false, unsupportedVersionException, null, null);

            if (!isInternalRequest)
                abortedSends.add(clientResponse);
            else if (clientRequest.apiKey() == ApiKeys.METADATA)
                metadataUpdater.handleFailedRequest(now, Optional.of(unsupportedVersionException));
        }
    }

    private void doSend(ClientRequest clientRequest, boolean isInternalRequest, long now, AbstractRequest request) {
        String destination = clientRequest.destination();
        RequestHeader header = clientRequest.makeHeader(request.version());
        if (log.isDebugEnabled()) {
            int latestClientVersion = clientRequest.apiKey().latestVersion();
            if (header.apiVersion() == latestClientVersion) {
                log.trace("Sending {} {} with correlation id {} to node {}", clientRequest.apiKey(), request,
                        clientRequest.correlationId(), destination);
            } else {
                log.debug("Using older server API v{} to send {} {} with correlation id {} to node {}",
                        header.apiVersion(), clientRequest.apiKey(), request, clientRequest.correlationId(), destination);
            }
        }
        Send send = request.toSend(destination, header);
        InFlightRequest inFlightRequest = new InFlightRequest(
                clientRequest,
                header,
                isInternalRequest,
                request,
                send,
                now);
        this.inFlightRequests.add(inFlightRequest);
        selector.send(send);
    }

    /**
     * Do actual reads and writes to sockets.
     *
     * @param timeout The maximum amount of time to wait (in ms) for responses if there are none immediately,
     *                must be non-negative. The actual timeout will be the minimum of timeout, request timeout and
     *                metadata timeout
     * @param now The current time in milliseconds
     * @return The list of responses received
     */
    @Override
    public List<ClientResponse> poll(long timeout, long now) {
        ensureActive();

        if (!abortedSends.isEmpty()) {
            // If there are aborted sends because of unsupported version exceptions or disconnects,
            // handle them immediately without waiting for Selector#poll.
            List<ClientResponse> responses = new ArrayList<>();
            handleAbortedSends(responses);
            completeResponses(responses);
            return responses;
        }

        long metadataTimeout = metadataUpdater.maybeUpdate(now);
        try {
            this.selector.poll(Utils.min(timeout, metadataTimeout, defaultRequestTimeoutMs));
        } catch (IOException e) {
            log.error("Unexpected error during I/O", e);
        }

        // process completed actions
        long updatedNow = this.time.milliseconds();
        List<ClientResponse> responses = new ArrayList<>();
        handleCompletedSends(responses, updatedNow);
        handleCompletedReceives(responses, updatedNow);
        handleDisconnections(responses, updatedNow);
        handleConnections();
        handleInitiateApiVersionRequests(updatedNow);
        handleTimedOutRequests(responses, updatedNow);
        completeResponses(responses);

        return responses;
    }

    private void completeResponses(List<ClientResponse> responses) {
        for (ClientResponse response : responses) {
            try {
                response.onComplete();
            } catch (Exception e) {
                log.error("Uncaught error in request completion:", e);
            }
        }
    }

    /**
     * Get the number of in-flight requests
     */
    @Override
    public int inFlightRequestCount() {
        return this.inFlightRequests.count();
    }

    @Override
    public boolean hasInFlightRequests() {
        return !this.inFlightRequests.isEmpty();
    }

    /**
     * Get the number of in-flight requests for a given node
     */
    @Override
    public int inFlightRequestCount(String node) {
        return this.inFlightRequests.count(node);
    }

    @Override
    public boolean hasInFlightRequests(String node) {
        return !this.inFlightRequests.isEmpty(node);
    }

    @Override
    public boolean hasReadyNodes(long now) {
        return connectionStates.hasReadyNodes(now);
    }

    /**
     * Interrupt the client if it is blocked waiting on I/O.
     */
    @Override
    public void wakeup() {
        this.selector.wakeup();
    }

    @Override
    public void initiateClose() {
        if (state.compareAndSet(State.ACTIVE, State.CLOSING)) {
            wakeup();
        }
    }

    @Override
    public boolean active() {
        return state.get() == State.ACTIVE;
    }

    private void ensureActive() {
        if (!active())
            throw new DisconnectException("NetworkClient is no longer active, state is " + state);
    }

    /**
     * Close the network client
     */
    @Override
    public void close() {
        state.compareAndSet(State.ACTIVE, State.CLOSING);
        if (state.compareAndSet(State.CLOSING, State.CLOSED)) {
            this.selector.close();
            this.metadataUpdater.close();
        } else {
            log.warn("Attempting to close NetworkClient that has already been closed.");
        }
    }

    /**
     * Choose the node with the fewest outstanding requests which is at least eligible for connection. This method will
     * prefer a node with an existing connection, but will potentially choose a node for which we don't yet have a
     * connection if all existing connections are in use. This method will never choose a node for which there is no
     * existing connection and from which we have disconnected within the reconnect backoff period, or an active
     * connection which is being throttled.
     *
     * @return The node with the fewest in-flight requests.
     */
    @Override
    public Node leastLoadedNode(long now) {
        List<Node> nodes = this.metadataUpdater.fetchNodes();
        if (nodes.isEmpty())
            throw new IllegalStateException("There are no nodes in the Kafka cluster");
        int inflight = Integer.MAX_VALUE;

        Node foundConnecting = null;
        Node foundCanConnect = null;
        Node foundReady = null;

        int offset = this.randOffset.nextInt(nodes.size());
        for (int i = 0; i < nodes.size(); i++) {
            int idx = (offset + i) % nodes.size();
            Node node = nodes.get(idx);
            if (canSendRequest(node.idString(), now)) {
                int currInflight = this.inFlightRequests.count(node.idString());
                if (currInflight == 0) {
                    // if we find an established connection with no in-flight requests we can stop right away
                    log.trace("Found least loaded node {} connected with no in-flight requests", node);
                    return node;
                } else if (currInflight < inflight) {
                    // otherwise if this is the best we have found so far, record that
                    inflight = currInflight;
                    foundReady = node;
                }
            } else if (connectionStates.isPreparingConnection(node.idString())) {
                foundConnecting = node;
            } else if (canConnect(node, now)) {
                foundCanConnect = node;
            } else {
                log.trace("Removing node {} from least loaded node selection since it is neither ready " +
                        "for sending or connecting", node);
            }
        }

        // We prefer established connections if possible. Otherwise, we will wait for connections
        // which are being established before connecting to new nodes.
        if (foundReady != null) {
            log.trace("Found least loaded node {} with {} inflight requests", foundReady, inflight);
            return foundReady;
        } else if (foundConnecting != null) {
            log.trace("Found least loaded connecting node {}", foundConnecting);
            return foundConnecting;
        } else if (foundCanConnect != null) {
            log.trace("Found least loaded node {} with no active connection", foundCanConnect);
            return foundCanConnect;
        } else {
            log.trace("Least loaded node selection failed to find an available node");
            return null;
        }
    }

    public static AbstractResponse parseResponse(ByteBuffer responseBuffer, RequestHeader requestHeader) {
        try {
            Struct responseStruct = parseStructMaybeUpdateThrottleTimeMetrics(responseBuffer, requestHeader, null, 0);
            return AbstractResponse.parseResponse(requestHeader.apiKey(), responseStruct,
                    requestHeader.apiVersion());
        } catch (BufferUnderflowException e) {
            throw new SchemaException("Buffer underflow while parsing response for request with header " + requestHeader, e);
        }
    }

    private static Struct parseStructMaybeUpdateThrottleTimeMetrics(ByteBuffer responseBuffer, RequestHeader requestHeader,
                                                                    Sensor throttleTimeSensor, long now) {
        ResponseHeader responseHeader = ResponseHeader.parse(responseBuffer,
            requestHeader.apiKey().responseHeaderVersion(requestHeader.apiVersion()));
        // Always expect the response version id to be the same as the request version id
        Struct responseBody = requestHeader.apiKey().parseResponse(requestHeader.apiVersion(), responseBuffer);
        correlate(requestHeader, responseHeader);
        if (throttleTimeSensor != null && responseBody.hasField(CommonFields.THROTTLE_TIME_MS))
            throttleTimeSensor.record(responseBody.get(CommonFields.THROTTLE_TIME_MS), now);
        return responseBody;
    }

    /**
     * Post process disconnection of a node
     *
     * @param responses The list of responses to update
     * @param nodeId Id of the node to be disconnected
     * @param now The current time
     * @param disconnectState The state of the disconnected channel
     */
    private void processDisconnection(List<ClientResponse> responses,
                                      String nodeId,
                                      long now,
                                      ChannelState disconnectState) {
        connectionStates.disconnected(nodeId, now);
        apiVersions.remove(nodeId);
        nodesNeedingApiVersionsFetch.remove(nodeId);
        switch (disconnectState.state()) {
            case AUTHENTICATION_FAILED:
                AuthenticationException exception = disconnectState.exception();
                connectionStates.authenticationFailed(nodeId, now, exception);
                log.error("Connection to node {} ({}) failed authentication due to: {}", nodeId,
                    disconnectState.remoteAddress(), exception.getMessage());
                break;
            case AUTHENTICATE:
                log.warn("Connection to node {} ({}) terminated during authentication. This may happen " +
                    "due to any of the following reasons: (1) Authentication failed due to invalid " +
                    "credentials with brokers older than 1.0.0, (2) Firewall blocking Kafka TLS " +
                    "traffic (eg it may only allow HTTPS traffic), (3) Transient network issue.",
                    nodeId, disconnectState.remoteAddress());
                break;
            case NOT_CONNECTED:
                log.warn("Connection to node {} ({}) could not be established. Broker may not be available.", nodeId, disconnectState.remoteAddress());
                break;
            default:
                break; // Disconnections in other states are logged at debug level in Selector
        }

        cancelInFlightRequests(nodeId, now, responses);
        metadataUpdater.handleServerDisconnect(now, nodeId, Optional.ofNullable(disconnectState.exception()));
    }

    /**
     * Iterate over all the inflight requests and expire any requests that have exceeded the configured requestTimeout.
     * The connection to the node associated with the request will be terminated and will be treated as a disconnection.
     *
     * @param responses The list of responses to update
     * @param now The current time
     */
    private void handleTimedOutRequests(List<ClientResponse> responses, long now) {
        List<String> nodeIds = this.inFlightRequests.nodesWithTimedOutRequests(now);
        for (String nodeId : nodeIds) {
            // close connection to the node
            this.selector.close(nodeId);
            log.debug("Disconnecting from node {} due to request timeout.", nodeId);
            processDisconnection(responses, nodeId, now, ChannelState.LOCAL_CLOSE);
        }
    }

    private void handleAbortedSends(List<ClientResponse> responses) {
        responses.addAll(abortedSends);
        abortedSends.clear();
    }

    /**
     * Handle any completed request send. In particular if no response is expected consider the request complete.
     *
     * @param responses The list of responses to update
     * @param now The current time
     */
    private void handleCompletedSends(List<ClientResponse> responses, long now) {
        // if no response is expected then when the send is completed, return it
        for (Send send : this.selector.completedSends()) {
            InFlightRequest request = this.inFlightRequests.lastSent(send.destination());
            if (!request.expectResponse) {
                this.inFlightRequests.completeLastSent(send.destination());
                responses.add(request.completed(null, now));
            }
        }
    }

    /**
     * If a response from a node includes a non-zero throttle delay and client-side throttling has been enabled for
     * the connection to the node, throttle the connection for the specified delay.
     *
     * @param response the response
     * @param apiVersion the API version of the response
     * @param nodeId the id of the node
     * @param now The current time
     */
    private void maybeThrottle(AbstractResponse response, short apiVersion, String nodeId, long now) {
        int throttleTimeMs = response.throttleTimeMs();
        if (throttleTimeMs > 0 && response.shouldClientThrottle(apiVersion)) {
            connectionStates.throttle(nodeId, now + throttleTimeMs);
            log.trace("Connection to node {} is throttled for {} ms until timestamp {}", nodeId, throttleTimeMs,
                      now + throttleTimeMs);
        }
    }

    /**
     * Handle any completed receives and update the response list with the responses received.
     *
     * @param responses The list of responses to update
     * @param now The current time
     */
    private void handleCompletedReceives(List<ClientResponse> responses, long now) {
        for (NetworkReceive receive : this.selector.completedReceives()) {
            String source = receive.source();
            InFlightRequest req = inFlightRequests.completeNext(source);
            Struct responseStruct = parseStructMaybeUpdateThrottleTimeMetrics(receive.payload(), req.header,
                throttleTimeSensor, now);
            if (log.isTraceEnabled()) {
                log.trace("Completed receive from node {} for {} with correlation id {}, received {}", req.destination,
                    req.header.apiKey(), req.header.correlationId(), responseStruct);
            }
            // If the received response includes a throttle delay, throttle the connection.
            AbstractResponse body = AbstractResponse.
                    parseResponse(req.header.apiKey(), responseStruct, req.header.apiVersion());
            maybeThrottle(body, req.header.apiVersion(), req.destination, now);
            if (req.isInternalRequest && body instanceof MetadataResponse)
                metadataUpdater.handleSuccessfulResponse(req.header, now, (MetadataResponse) body);
            else if (req.isInternalRequest && body instanceof ApiVersionsResponse)
                handleApiVersionsResponse(responses, req, now, (ApiVersionsResponse) body);
            else
                responses.add(req.completed(body, now));
        }
    }

    private void handleApiVersionsResponse(List<ClientResponse> responses,
                                           InFlightRequest req, long now, ApiVersionsResponse apiVersionsResponse) {
        final String node = req.destination;
        if (apiVersionsResponse.data.errorCode() != Errors.NONE.code()) {
            if (req.request.version() == 0 || apiVersionsResponse.data.errorCode() != Errors.UNSUPPORTED_VERSION.code()) {
                log.warn("Received error {} from node {} when making an ApiVersionsRequest with correlation id {}. Disconnecting.",
                        Errors.forCode(apiVersionsResponse.data.errorCode()), node, req.header.correlationId());
                this.selector.close(node);
                processDisconnection(responses, node, now, ChannelState.LOCAL_CLOSE);
            } else {
                // Starting from Apache Kafka 2.4, ApiKeys field is populated with the supported versions of
                // the ApiVersionsRequest when an UNSUPPORTED_VERSION error is returned.
                // If not provided, the client falls back to version 0.
                short maxApiVersion = 0;
                if (apiVersionsResponse.data.apiKeys().size() > 0) {
                    ApiVersionsResponseKey apiVersion = apiVersionsResponse.data.apiKeys().find(ApiKeys.API_VERSIONS.id);
                    if (apiVersion != null) {
                        maxApiVersion = apiVersion.maxVersion();
                    }
                }
                nodesNeedingApiVersionsFetch.put(node, new ApiVersionsRequest.Builder(maxApiVersion));
            }
            return;
        }
        NodeApiVersions nodeVersionInfo = new NodeApiVersions(apiVersionsResponse.data.apiKeys());
        apiVersions.update(node, nodeVersionInfo);
        this.connectionStates.ready(node);
        log.debug("Recorded API versions for node {}: {}", node, nodeVersionInfo);
    }

    /**
     * Handle any disconnected connections
     *
     * @param responses The list of responses that completed with the disconnection
     * @param now The current time
     */
    private void handleDisconnections(List<ClientResponse> responses, long now) {
        for (Map.Entry<String, ChannelState> entry : this.selector.disconnected().entrySet()) {
            String node = entry.getKey();
            log.debug("Node {} disconnected.", node);
            processDisconnection(responses, node, now, entry.getValue());
        }
    }

    /**
     * Record any newly completed connections
     */
    private void handleConnections() {
        for (String node : this.selector.connected()) {
            // We are now connected.  Note that we might not still be able to send requests. For instance,
            // if SSL is enabled, the SSL handshake happens after the connection is established.
            // Therefore, it is still necessary to check isChannelReady before attempting to send on this
            // connection.
            if (discoverBrokerVersions) {
                this.connectionStates.checkingApiVersions(node);
                nodesNeedingApiVersionsFetch.put(node, new ApiVersionsRequest.Builder());
                log.debug("Completed connection to node {}. Fetching API versions.", node);
            } else {
                this.connectionStates.ready(node);
                log.debug("Completed connection to node {}. Ready.", node);
            }
        }
    }

    private void handleInitiateApiVersionRequests(long now) {
        Iterator<Map.Entry<String, ApiVersionsRequest.Builder>> iter = nodesNeedingApiVersionsFetch.entrySet().iterator();
        while (iter.hasNext()) {
            Map.Entry<String, ApiVersionsRequest.Builder> entry = iter.next();
            String node = entry.getKey();
            if (selector.isChannelReady(node) && inFlightRequests.canSendMore(node)) {
                log.debug("Initiating API versions fetch from node {}.", node);
                ApiVersionsRequest.Builder apiVersionRequestBuilder = entry.getValue();
                ClientRequest clientRequest = newClientRequest(node, apiVersionRequestBuilder, now, true);
                doSend(clientRequest, true, now);
                iter.remove();
            }
        }
    }

    /**
     * Validate that the response corresponds to the request we expect or else explode
     */
    private static void correlate(RequestHeader requestHeader, ResponseHeader responseHeader) {
        if (requestHeader.correlationId() != responseHeader.correlationId()) {
            if (SaslClientAuthenticator.isReserved(requestHeader.correlationId())
                    && !SaslClientAuthenticator.isReserved(responseHeader.correlationId()))
                throw new SchemaException("the response is unrelated to Sasl request since its correlation id is " + responseHeader.correlationId()
                    + " and the reserved range for Sasl request is [ "
                    + SaslClientAuthenticator.MIN_RESERVED_CORRELATION_ID + "," + SaslClientAuthenticator.MAX_RESERVED_CORRELATION_ID + "]");
            throw new IllegalStateException("Correlation id for response (" + responseHeader.correlationId()
                    + ") does not match request (" + requestHeader.correlationId() + "), request header: " + requestHeader);
        }
    }

    /**
     * Initiate a connection to the given node
     * @param node the node to connect to
     * @param now current time in epoch milliseconds
     */
    private void initiateConnect(Node node, long now) {
        String nodeConnectionId = node.idString();
        try {
            connectionStates.connecting(nodeConnectionId, now, node.host(), clientDnsLookup);
            InetAddress address = connectionStates.currentAddress(nodeConnectionId);
            log.debug("Initiating connection to node {} using address {}", node, address);
            selector.connect(nodeConnectionId,
                    new InetSocketAddress(address, node.port()),
                    this.socketSendBuffer,
                    this.socketReceiveBuffer);
        } catch (IOException e) {
            log.warn("Error connecting to node {}", node, e);
            // Attempt failed, we'll try again after the backoff
            connectionStates.disconnected(nodeConnectionId, now);
            // Notify metadata updater of the connection failure
            metadataUpdater.handleServerDisconnect(now, nodeConnectionId, Optional.empty());
        }
    }

    class DefaultMetadataUpdater implements MetadataUpdater {

        /* the current cluster metadata */
        private final Metadata metadata;

        // Defined if there is a request in progress, null otherwise
        private InProgressData inProgress;

        DefaultMetadataUpdater(Metadata metadata) {
            this.metadata = metadata;
            this.inProgress = null;
        }

        @Override
        public List<Node> fetchNodes() {
            return metadata.fetch().nodes();
        }

        @Override
        public boolean isUpdateDue(long now) {
            return !hasFetchInProgress() && this.metadata.timeToNextUpdate(now) == 0;
        }

        private boolean hasFetchInProgress() {
            return inProgress != null;
        }

        @Override
        public long maybeUpdate(long now) {
            // should we update our metadata?
            long timeToNextMetadataUpdate = metadata.timeToNextUpdate(now);
            long waitForMetadataFetch = hasFetchInProgress() ? defaultRequestTimeoutMs : 0;

            long metadataTimeout = Math.max(timeToNextMetadataUpdate, waitForMetadataFetch);
            if (metadataTimeout > 0) {
                return metadataTimeout;
            }

            // Beware that the behavior of this method and the computation of timeouts for poll() are
            // highly dependent on the behavior of leastLoadedNode.
            Node node = leastLoadedNode(now);
            if (node == null) {
                log.debug("Give up sending metadata request since no node is available");
                return reconnectBackoffMs;
            }

            return maybeUpdate(now, node);
        }

        @Override
        public void handleServerDisconnect(long now, String destinationId, Optional<AuthenticationException> maybeFatalException) {
            Cluster cluster = metadata.fetch();
            // 'processDisconnection' generates warnings for misconfigured bootstrap server configuration
            // resulting in 'Connection Refused' and misconfigured security resulting in authentication failures.
            // The warning below handles the case where a connection to a broker was established, but was disconnected
            // before metadata could be obtained.
            if (cluster.isBootstrapConfigured()) {
                int nodeId = Integer.parseInt(destinationId);
                Node node = cluster.nodeById(nodeId);
                if (node != null)
                    log.warn("Bootstrap broker {} disconnected", node);
            }

            // If we have a disconnect while an update is due, we treat it as a failed update
            // so that we can backoff properly
            if (isUpdateDue(now))
                handleFailedRequest(now, Optional.empty());

            maybeFatalException.ifPresent(metadata::fatalError);

            // The disconnect may be the result of stale metadata, so request an update
            metadata.requestUpdate();
        }

        @Override
        public void handleFailedRequest(long now, Optional<KafkaException> maybeFatalException) {
            maybeFatalException.ifPresent(metadata::fatalError);
            metadata.failedUpdate(now);
            inProgress = null;
        }

        @Override
        public void handleSuccessfulResponse(RequestHeader requestHeader, long now, MetadataResponse response) {
            // If any partition has leader with missing listeners, log up to ten of these partitions
            // for diagnosing broker configuration issues.
            // This could be a transient issue if listeners were added dynamically to brokers.
            List<TopicPartition> missingListenerPartitions = response.topicMetadata().stream().flatMap(topicMetadata ->
                topicMetadata.partitionMetadata().stream()
                    .filter(partitionMetadata -> partitionMetadata.error == Errors.LISTENER_NOT_FOUND)
                    .map(partitionMetadata -> new TopicPartition(topicMetadata.topic(), partitionMetadata.partition())))
                .collect(Collectors.toList());
            if (!missingListenerPartitions.isEmpty()) {
                int count = missingListenerPartitions.size();
                log.warn("{} partitions have leader brokers without a matching listener, including {}",
                        count, missingListenerPartitions.subList(0, Math.min(10, count)));
            }

            // Check if any topic's metadata failed to get updated
            Map<String, Errors> errors = response.errors();
            if (!errors.isEmpty())
                log.warn("Error while fetching metadata with correlation id {} : {}", requestHeader.correlationId(), errors);

            // Don't update the cluster if there are no valid nodes...the topic we want may still be in the process of being
            // created which means we will get errors and no nodes until it exists
            if (response.brokers().isEmpty()) {
                log.trace("Ignoring empty metadata response with correlation id {}.", requestHeader.correlationId());
                this.metadata.failedUpdate(now);
            } else {
                this.metadata.update(inProgress.requestVersion, response, inProgress.isPartialUpdate, now);
            }

            inProgress = null;
        }

        @Override
        public void close() {
            this.metadata.close();
        }

        /**
         * Return true if there's at least one connection establishment is currently underway
         */
        private boolean isAnyNodeConnecting() {
            for (Node node : fetchNodes()) {
                if (connectionStates.isConnecting(node.idString())) {
                    return true;
                }
            }
            return false;
        }

        /**
         * Add a metadata request to the list of sends if we can make one
         */
        private long maybeUpdate(long now, Node node) {
            String nodeConnectionId = node.idString();

            if (canSendRequest(nodeConnectionId, now)) {
                Metadata.MetadataRequestAndVersion requestAndVersion = metadata.newMetadataRequestAndVersion(now);
                MetadataRequest.Builder metadataRequest = requestAndVersion.requestBuilder;
                log.debug("Sending metadata request {} to node {}", metadataRequest, node);
                sendInternalMetadataRequest(metadataRequest, nodeConnectionId, now);
                inProgress = new InProgressData(requestAndVersion.requestVersion, requestAndVersion.isPartialUpdate);
                return defaultRequestTimeoutMs;
            }

            // If there's any connection establishment underway, wait until it completes. This prevents
            // the client from unnecessarily connecting to additional nodes while a previous connection
            // attempt has not been completed.
            if (isAnyNodeConnecting()) {
                // Strictly the timeout we should return here is "connect timeout", but as we don't
                // have such application level configuration, using reconnect backoff instead.
                return reconnectBackoffMs;
            }

            if (connectionStates.canConnect(nodeConnectionId, now)) {
                // We don't have a connection to this node right now, make one
                log.debug("Initialize connection to node {} for sending metadata request", node);
                initiateConnect(node, now);
                return reconnectBackoffMs;
            }

            // connected, but can't send more OR connecting
            // In either case, we just need to wait for a network event to let us know the selected
            // connection might be usable again.
            return Long.MAX_VALUE;
        }

        public class InProgressData {
            public final int requestVersion;
            public final boolean isPartialUpdate;

            private InProgressData(int requestVersion, boolean isPartialUpdate) {
                this.requestVersion = requestVersion;
                this.isPartialUpdate = isPartialUpdate;
            }
        };

    }

    @Override
    public ClientRequest newClientRequest(String nodeId,
                                          AbstractRequest.Builder<?> requestBuilder,
                                          long createdTimeMs,
                                          boolean expectResponse) {
        return newClientRequest(nodeId, requestBuilder, createdTimeMs, expectResponse, defaultRequestTimeoutMs, null);
    }

    // visible for testing
    int nextCorrelationId() {
        if (SaslClientAuthenticator.isReserved(correlation)) {
            // the numeric overflow is fine as negative values is acceptable
            correlation = SaslClientAuthenticator.MAX_RESERVED_CORRELATION_ID + 1;
        }
        return correlation++;
    }

    @Override
    public ClientRequest newClientRequest(String nodeId,
                                          AbstractRequest.Builder<?> requestBuilder,
                                          long createdTimeMs,
                                          boolean expectResponse,
                                          int requestTimeoutMs,
                                          RequestCompletionHandler callback) {
        return new ClientRequest(nodeId, requestBuilder, nextCorrelationId(), clientId, createdTimeMs, expectResponse,
                requestTimeoutMs, callback);
    }

    public boolean discoverBrokerVersions() {
        return discoverBrokerVersions;
    }

    static class InFlightRequest {
        final RequestHeader header;
        final String destination;
        final RequestCompletionHandler callback;
        final boolean expectResponse;
        final AbstractRequest request;
        final boolean isInternalRequest; // used to flag requests which are initiated internally by NetworkClient
        final Send send;
        final long sendTimeMs;
        final long createdTimeMs;
        final long requestTimeoutMs;

        public InFlightRequest(ClientRequest clientRequest,
                               RequestHeader header,
                               boolean isInternalRequest,
                               AbstractRequest request,
                               Send send,
                               long sendTimeMs) {
            this(header,
                 clientRequest.requestTimeoutMs(),
                 clientRequest.createdTimeMs(),
                 clientRequest.destination(),
                 clientRequest.callback(),
                 clientRequest.expectResponse(),
                 isInternalRequest,
                 request,
                 send,
                 sendTimeMs);
        }

        public InFlightRequest(RequestHeader header,
                               int requestTimeoutMs,
                               long createdTimeMs,
                               String destination,
                               RequestCompletionHandler callback,
                               boolean expectResponse,
                               boolean isInternalRequest,
                               AbstractRequest request,
                               Send send,
                               long sendTimeMs) {
            this.header = header;
            this.requestTimeoutMs = requestTimeoutMs;
            this.createdTimeMs = createdTimeMs;
            this.destination = destination;
            this.callback = callback;
            this.expectResponse = expectResponse;
            this.isInternalRequest = isInternalRequest;
            this.request = request;
            this.send = send;
            this.sendTimeMs = sendTimeMs;
        }

        public ClientResponse completed(AbstractResponse response, long timeMs) {
            return new ClientResponse(header, callback, destination, createdTimeMs, timeMs,
                    false, null, null, response);
        }

        public ClientResponse disconnected(long timeMs, AuthenticationException authenticationException) {
            return new ClientResponse(header, callback, destination, createdTimeMs, timeMs,
                    true, null, authenticationException, null);
        }

        @Override
        public String toString() {
            return "InFlightRequest(header=" + header +
                    ", destination=" + destination +
                    ", expectResponse=" + expectResponse +
                    ", createdTimeMs=" + createdTimeMs +
                    ", sendTimeMs=" + sendTimeMs +
                    ", isInternalRequest=" + isInternalRequest +
                    ", request=" + request +
                    ", callback=" + callback +
                    ", send=" + send + ")";
        }
    }

}


text:
using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Threading.Tasks;
using Windows.Foundation;
using Windows.Foundation.Collections;
using Windows.Storage;
using Windows.System;
using Windows.UI.ApplicationSettings;
using Windows.UI.ViewManagement;
using Windows.UI.Xaml;
using Windows.UI.Xaml.Controls;
using Windows.UI.Xaml.Controls.Primitives;
using Windows.UI.Xaml.Data;
using Windows.UI.Xaml.Input;
using Windows.UI.Xaml.Media;
using Windows.UI.Xaml.Navigation;

// The Blank Page item template is documented at http://go.microsoft.com/fwlink/?LinkId=234238

namespace DrumKit
{
    /// <summary>
    /// An empty page that can be used on its own or navigated to within a Frame.
    /// </summary>
    public sealed partial class MainPage : Page
    {
        #region Fields: Private members
        private Dictionary<string, DrumPlayUI> DrumUIs { get; set; }
        private Dictionary<VirtualKey, string> Keymap { get; set; }
        private int CurrentLayout { get; set; }

        #endregion

        #region Initialization
        /// <summary>
        /// Creates a new instance of MainPage.
        /// </summary>
        public MainPage()
        {
            // Create private members
            this.DrumUIs = new Dictionary<string, DrumPlayUI>();
            this.Keymap = new Dictionary<VirtualKey, string>();
            CurrentLayout = 0;

            // Initialize page
            this.InitializeComponent();
            this.SizeChanged += MainPage_SizeChanged;
            this.Loaded += MainPage_Loaded;

            SettingsPane.GetForCurrentView().CommandsRequested += MainPage_CommandsRequested;

            // Initialize drums
            this.InitializeDrums();
        }

        /// <summary>
        /// Sets up a single drum
        /// </summary>
        /// <param name="drum">The drum.</param>
        private void InitializeDrum(Drum drum)
        {
            // Create object
            DrumPlayUI d = new DrumPlayUI(drum);

            // Set up callbacks
            d.Hit += HandlerDrumPointerPressed;

            // Add to lists
            canvas.Children.Add(d);
            this.DrumUIs.Add(drum.Id, d);
        }

        /// <summary>
        /// Sets up the drums.
        /// </summary>
        private void InitializeDrums()
        {
            // Clear previous stuff if any
            this.DrumUIs.Clear();

            // Load drums
            foreach (var i in DataController.CurrentDrumkit.DrumsList)
                InitializeDrum(i);

            UpdateDrumConfig();
        }

        /// <summary>
        /// Sets up the drum configurations
        /// </summary>
        private void UpdateDrumConfig()
        {
            this.Keymap.Clear();

            // Load drum configurations
            foreach (var i in DataController.CurrentConfig.DrumsList)
            {
                // Unload if disabled
                if (!i.IsEnabled)
                {
                    canvas.Children.Remove(this.DrumUIs[i.TargetId]);
                    this.DrumUIs.Remove(i.TargetId);
                }

                else
                {
                    // Set drum key
                    this.DrumUIs[i.TargetId].KeyString = UIHelper.GetPrettifiedVKeyName(i.Key);
                    this.DrumUIs[i.TargetId].IsKeyVisible = DataController.Settings.ShowKeyBindings;

                    // Keyboard mapping
                    if (!Keymap.ContainsKey(i.Key))
                        Keymap.Add(i.Key, i.TargetId);
                }
            }
        }

        #endregion

        #region UI: Settings charm
        /// <summary>
        /// Triggered when the settings pane requests commands/
        /// </summary>
        void MainPage_CommandsRequested(SettingsPane sender, SettingsPaneCommandsRequestedEventArgs args)
        {
            args.Request.ApplicationCommands.Clear();
            SettingsCommand general = new SettingsCommand("general", "General", new Windows.UI.Popups.UICommandInvokedHandler(MainPage_SettingsCommand));
            SettingsCommand drumkits = new SettingsCommand("drumkits", "Manage drumkits", new Windows.UI.Popups.UICommandInvokedHandler(MainPage_SettingsCommand));
            SettingsCommand drums = new SettingsCommand("drums", "Manage drums", new Windows.UI.Popups.UICommandInvokedHandler(MainPage_SettingsCommand));
            SettingsCommand layouts = new SettingsCommand("layouts", "Layouts", new Windows.UI.Popups.UICommandInvokedHandler(MainPage_SettingsCommand));

            args.Request.ApplicationCommands.Add(general);
            args.Request.ApplicationCommands.Add(drumkits);
            args.Request.ApplicationCommands.Add(drums);
            args.Request.ApplicationCommands.Add(layouts);
        }

        /// <summary>
        /// Handles the settings charms
        /// </summary>
        void MainPage_SettingsCommand(Windows.UI.Popups.IUICommand command)
        {
            Frame.Navigate(typeof(SettingsPage), command.Id);
        }

        #endregion

        #region Layouts

        /// <summary>
        /// Figures out which is the best layout available, and uses it.
        /// </summary>
        private int PickBestLayout()
        {
            // Smaller index is better
            int[] picks = {-1, -1, -1, -1, -1, -1};

            // Get current layout
            var view = UIHelper.ToDrumkitLayoutView(Windows.UI.ViewManagement.ApplicationView.Value);

            // Find best option
            for (int index = 0; index < DataController.CurrentLayouts.Items.Count; index++ )
            {
                var i = DataController.CurrentLayouts.Items[index];

                bool isSame = (i.TargetView == view);
                bool contains = (i.TargetView & view) > 0;
                bool all = i.TargetView == DrumkitLayoutTargetView.All;

                if (i.IsDefault)
                {
                    if (isSame) picks[0] = index;
                    if (contains) picks[1] = index;
                    if (all) picks[2] = index;
                }

                else
                {
                    if (isSame) picks[3] = index;
                    if (contains) picks[4] = index;
                    if (all) picks[5] = index;
                }
            }

            // Return first value different than -1, or 0
            foreach (var i in picks)
                if (i != -1) return i;

            return 0;
        }

        /// <summary>
        /// Sets up the layout
        /// </summary>
        private void ReloadLayout()
        {
            // Get current size
            double w = canvas.ActualWidth;
            double h = canvas.ActualHeight;

            if (double.IsNaN(w) || double.IsNaN(h) || double.IsInfinity(w) || double.IsInfinity(h))
                return;

            // Pick a layout
            this.CurrentLayout = PickBestLayout();

            Log.Write("Layout change: picked_layout={0}, w={1}, h={2}", this.CurrentLayout, w, h);

            // Apply layout
            foreach (var i in DataController.CurrentLayouts.Items[CurrentLayout].Drums)
            {
                if (!DrumUIs.ContainsKey(i.Key))
                    continue;

                // Set angle
                DrumUIs[i.Key].Angle = i.Value.Angle;

                // Set scale
                DrumUIs[i.Key].Width = i.Value.Size * w;
                DrumUIs[i.Key].Height = i.Value.Size * w;

                // Set position
                Canvas.SetLeft(DrumUIs[i.Key], w * i.Value.X);
                Canvas.SetTop(DrumUIs[i.Key], h * i.Value.Y);
            }
        }

        #endregion

        #region UI: Drums
        /// <summary>
        /// Handles drum hit using mouse/finger
        /// </summary>
        void HandlerDrumPointerPressed(object sender, PointerRoutedEventArgs e)
        {
            var button = sender as DrumPlayUI;

            if (button != null)
                this.HandlerDrumClickedCommon(button.DrumID);
        }

        /// <summary>
        /// Handles drum hit using keyboard
        /// </summary>
        public void HandlerKeyDown(Windows.UI.Core.CoreWindow sender, Windows.UI.Core.KeyEventArgs args)
        {
            string drum;

            // Key repeat or something
            if (args.KeyStatus.WasKeyDown)
                return;

            // If key in key map, perform "click"
            if (this.Keymap.TryGetValue(args.VirtualKey, out drum))
                HandlerDrumClickedCommon(drum);
        }

        /// <summary>
        /// Handles drum hit.
        /// </summary>
        private void HandlerDrumClickedCommon(string drum_id)
        {
            try
            {
                DataController.PlaySound(drum_id, 0);

                if (DataController.Settings.Animations)
                    this.DrumUIs[drum_id].PerformHit();
            }

            catch (Exception ex)
            {
                Log.Error("Error at playback!!! Drum id: {0}", drum_id);
                Log.Except(ex);
            }
        }

        #endregion

        #region UI: Page events
        /// <summary>
        /// Handles page size change event.
        /// </summary>
        void MainPage_SizeChanged(object sender, SizeChangedEventArgs e)
        {
            this.ReloadLayout();
            this.ReloadLayout();
        }

        /// <summary>
        /// Handles page load event.
        /// </summary>
        void MainPage_Loaded(object sender, RoutedEventArgs e)
        {
            // Set layout
            this.ReloadLayout();

            // Set toggles
            buttonAnimations.IsChecked = DataController.Settings.Animations;
            buttonKeys.IsChecked = DataController.Settings.ShowKeyBindings;
        }

        /// <summary>
        /// Invoked when this page is about to be displayed in a Frame.
        /// </summary>
        /// <param name="e">Event data that describes how this page was reached.  The Parameter
        /// property is typically used to configure the page.</param>
        protected override void OnNavigatedTo(NavigationEventArgs e)
        {
            Log.Write("Navigated to main page.");

            Window.Current.CoreWindow.KeyDown += this.HandlerKeyDown;
        }

        /// <summary>
        /// Invoked when the page is about to be destroyed.
        /// </summary>
        protected override void OnNavigatedFrom(NavigationEventArgs e)
        {
            Window.Current.CoreWindow.KeyDown -= this.HandlerKeyDown;
        }

        #endregion

        #region UI: Buttons
        /// <summary>
        /// Handles the edit button, going into the editor.
        /// </summary>
        private void ButtonEditMode_Click(object sender, RoutedEventArgs e)
        {
            Frame.Navigate(typeof(EditorPage), DataController.CurrentLayouts.Items[CurrentLayout]);
        }

        /// <summary>
        /// Handles the animations enabled toggle button.
        /// </summary>
        private void buttonAnimations_Click(object sender, RoutedEventArgs e)
        {
            var button = sender as ToggleButton;
            if (button == null) return;

            bool isChecked = (button.IsChecked.HasValue && button.IsChecked.Value);

            // Fix togglebuton style bug
            VisualStateManager.GoToState(button, isChecked ? "Checked" : "Unchecked", false);

            // Change animation setting
            DataController.Settings.Animations = isChecked;

            // Save modified setting
            DataController.SaveSettings();
        }

        /// <summary>
        /// Handles the 'show keys' toggle button
        /// </summary>
        private void buttonKeys_Click(object sender, RoutedEventArgs e)
        {
            var button = sender as ToggleButton;
            if (button == null) return;

            bool isChecked = (button.IsChecked.HasValue && button.IsChecked.Value);

            // Fix togglebuton style bug
            VisualStateManager.GoToState(button, isChecked ? "Checked" : "Unchecked", false);

            // Change setting
            DataController.Settings.ShowKeyBindings = isChecked;

            // Update UI
            UpdateDrumConfig();

            // Save modified setting
            DataController.SaveSettings();
        }

        /// <summary>
        /// Goes to application settings.
        /// </summary>
        private void ButtonSettings_Click(object sender, RoutedEventArgs e)
        {
            Frame.Navigate(typeof(SettingsPage), this);
        }

        #endregion
    }
}

redpajama_set_name: RedPajamaGithub


text:
<?php

namespace WikiRenderer\Markup\Trac;

use WikiRenderer\Generator\BlockListInterface;

/**
 * Parse a list block.
 */
class WikiList extends \WikiRenderer\Block
{
    public $type = 'list';
    protected $regexp = "/^(\s*)(\-|\*|\d\.)\s*(.*)/";

    /**
     * @var BlockListInterface[]
     */
    protected $generatorStack = array();

    /**
     * @var array of array(indent len, 'o' or 'u');
     */
    protected $indentStack = array();

    public function isStarting($line)
    {
        return $this->isAccepting($line);
    }

    public function isAccepting($string)
    {
        $this->sameItem = false;
        if (!preg_match($this->regexp, $string, $this->_detectMatch)) {
            if (!count($this->indentStack)) {
                return false;
            }
            if (!preg_match("/^(\s*)([\*#\-\!\| \t>;<=].*)$/", $string, $this->_detectMatch)) {
                return false;
            }
            if (strlen($this->_detectMatch[1]) < end($this->indentStack)[0]) {
                return false;
            }
            $this->sameItem = true;

            return true;
        }

        // if we are already here and the indentation length is lower
        // than the one of the first detected item so it is a new list.
        // we should start an other block
        if (count($this->indentStack) && $this->indentStack[0][0] > strlen($this->_detectMatch[1])) {
            return false;
        }

        // if this is not the same list type at the first level,
        // we should start an other block
        $type = $this->getItemType($this->_detectMatch[2]);
        if (count($this->indentStack) == 1 &&
            $this->indentStack[0][0] == strlen($this->_detectMatch[1]) &&
            $this->indentStack[0][1] != $type) {
            return false;
        }

        return true;
    }

    protected function getItemType($type)
    {
        return ($type == '*' || $type == '-') ? 'u' : 'o';
    }

    public function open()
    {
        $type = $this->getItemType($this->_detectMatch[2]);
        $this->generatorStack = array($this->generator);
        $this->indentStack = array(array(strlen($this->_detectMatch[1]), $type));

        if ($type == 'u') {
            $this->generator->setListType(BlockListInterface::UNORDERED_LIST);
        } else {
            $this->generator->setListType(BlockListInterface::ORDERED_LIST);
        }
    }

    public function close($reason)
    {
        $this->generatorStack = array();
        $this->indentStack = array();

        return parent::close($reason);
    }

    protected function _createList($type)
    {
        $generator = $this->documentGenerator->getBlockGenerator('list');
        if ($type == 'u') {
            $generator->setListType(BlockListInterface::UNORDERED_LIST);
        } else {
            $generator->setListType(BlockListInterface::ORDERED_LIST);
        }
        $generator->createItem();

        return $generator;
    }

    public function validateLine()
    {
        if ($this->sameItem) {
            $last = $this->generatorStack[count($this->generatorStack) - 1];
            $last->addContentToItem($this->parseInlineContent($this->_detectMatch[2]));

            return;
        }

        $t = end($this->indentStack);
        $newLen = strlen($this->_detectMatch[1]);
        $d = $t[0] - $newLen;
        $type = $this->getItemType($this->_detectMatch[2]);

        if ($d > 0) {
            // we pop off the list of nested list
            for ($i = count($this->indentStack) - 1; $i >= 0; --$i) {
                if ($this->indentStack[$i][0] <= $newLen) {
                    break;
                }
                array_pop($this->indentStack);
                array_pop($this->generatorStack);
            }

            // the new item is not of the same type
            // we should close the current sub list and create a new one
            $t = end($this->indentStack);
            if ($t[1] != $type) {
                array_pop($this->generatorStack);
                array_pop($this->indentStack);
                $this->indentStack[] = array($t[0],$type);
                $generator = $this->_createList($type);
                $generator->addContentToItem($this->parseInlineContent($this->_detectMatch[3]));
                $last = $this->generatorStack[count($this->generatorStack) - 1];
                $last->addContentToItem($generator);
                $this->generatorStack[] = $generator;

                return;
            }
        } elseif ($d < 0) { // we have a new nested list

            $generator = $this->_createList($type);
            $generator->addContentToItem($this->parseInlineContent($this->_detectMatch[3]));

            $last = $this->generatorStack[count($this->generatorStack) - 1];
            $last->addContentToItem($generator);
            $this->generatorStack[] = $generator;
            $this->indentStack[] = array(strlen($this->_detectMatch[1]), $type);

            return;
        } else {
            // the new item is not of the same type
            // we should close the current sub list and create a new one
            if ($t[1] != $type) {
                array_pop($this->generatorStack);
                array_pop($this->indentStack);
                $this->indentStack[] = array($t[0], $type);
                $generator = $this->_createList($type);
                $generator->addContentToItem($this->parseInlineContent($this->_detectMatch[3]));
                $last = $this->generatorStack[count($this->generatorStack) - 1];
                $last->addContentToItem($generator);
                $this->generatorStack[] = $generator;

                return;
            }
        }

        $last = $this->generatorStack[count($this->generatorStack) - 1];
        $last->createItem();
        $last->addContentToItem($this->parseInlineContent($this->_detectMatch[3]));
    }
}


text:
[![npm version](https://badge.fury.io/js/%40angular%2Fflex-layout.svg)](https://www.npmjs.com/package/%40angular%2Fflex-layout)
[![Build Status](https://travis-ci.org/angular/flex-layout.svg?branch=master)](https://travis-ci.org/angular/flex-layout)
[![Gitter](https://badges.gitter.im/angular/flex-layout.svg)](https://gitter.im/angular/flex-layout))

Angular Flex Layout provides a sophisticated layout API using FlexBox CSS + mediaQuery.
This module provides Angular (v4.x and higher) developers with component layout features using a
custom Layout API, mediaQuery observables,and DOM flexbox-2016 css.

The Layout engine intelligently automates the process of applying appropriate FlexBox CSS to
browser view hierarchies. This automation also addresses many of the complexities and workarounds
encountered with the traditional, manual, CSS-only application of Flexbox CSS.

##### Responsive API
The **real** power of Flex Layout, however, is its **responsive** engine.

The [Responsive API](https://github.com/angular/flex-layout/wiki/API-Overview#responsive-features)
enables developers to easily specify different layouts, sizing, visibilities for different
viewport sizes and display devices.

---

#### Deprecated support

Beta.9 provides support for Angular 4.x, AOT, Universal builds. This also means that Beta.9 (or higher) will not support Angular 2.x.

----

### Updated CHANGELOG

Please consult our [CHANGELOG](https://github.com/angular/flex-layout-builds/blob/master/CHANGELOG.md) for latest changes and new features.

### Why choose Flex-Layout

While other Flexbox CSS libraries are implementations of:

* pure CSS-only implementations, or
* the JS+CSS Stylesheets implementation of Angular Material v1.x Layouts.

Angular Flex Layout - in contrast - is a pure-Typescript UI Layout engine with an implementation that:

*  uses HTML attributes (aka Layout API) to specify the layout configurations
*  is currently only available for Angular (v4.1 or higher) Applications.
*  is independent of Angular Material (v1 or v2).
*  requires no external stylesheets.
*  requires Angular v4.1 or higher.

### Browser Support

### NPM Installs

Playing with the nightly changes in [master](https://github.com/angular/flex-layout/tree/master) is also possible

```bash
npm install https://github.com/angular/flex-layout-builds.git
````

Or you can easily install the latest release from NPM using:

```bash
npm install @angular/flex-layout@latest --save
```

### Quick Links

*  [Wiki Docs](https://github.com/angular/flex-layout/wiki)
*  [License: MIT](https://raw.githubusercontent.com/angular/flex-layout-builds/master/LICENSE)

Developer Guides

* [NPM Installs](https://github.com/angular/flex-layout/wiki/NPM-Installs)
* [Developer Setup](https://github.com/angular/flex-layout/wiki/Developer-Setup)
* [Builds + Fast Starts](https://github.com/angular/flex-layout/wiki/Fast-Starts)
* [Using Angular CLI](https://github.com/angular/flex-layout/wiki/Using-Angular-CLI)
* [WebPack Configuration](https://github.com/angular/flex-layout/wiki/Webpack-Configuration)

Documentation

*  [Static API](https://github.com/angular/flex-layout/wiki/Declarative-API-Overview)
*  [Responsive API](https://github.com/angular/flex-layout/wiki/Responsive-API)
*  [API Documentation](https://github.com/angular/flex-layout/wiki/API-Documentation)
*  [Custom Breakpoints](https://github.com/angular/flex-layout/wiki/Custom-Breakpoints)
*  [Best Performance](https://github.com/angular/flex-layout/wiki/Best-Performance)

Demos

*  [Live Online](https://tburleson-layouts-demos.firebaseapp.com/)
*  [Source Code](https://github.com/angular/flex-layout/blob/master/src/demo-app/app/demo-app-module.ts)

Templates

*  [Plunkr Template](https://plnkr.co/edit/h8hzyoEyqdCXmTBA7DfK?p=preview)

Quick Links

*  [Gitter Chat](https://gitter.im/angular/flex-layout)
*  [Discussion Forum](https://groups.google.com/forum/#!forum/angular-flex-layout)

### License

License: MIT

The sources for this package are in the [Flex-Layout](https://github.com/angular/flex-layout) repository.

Please file issues and pull requests against that repo.


text:
Spacer helps you manage your microservice dependencies.

## Synopsis

First, create a file called `Spacerfile` with the following content:
```
poga/spacer/examples/counter
poga/spacer/examples/hello
```

Now we can boot them up and setup a easy-to-use proxy with one command:
```
$ spacer
Cloning git@github.com:poga/spacer.git into services/poga/spacer ...
Building services/poga/spacer/examples/counter/docker-compose.yml ...
...
...
Successfully built c90b052fff60

Starting services/poga/spacer/examples/counter/docker-compose.yml ...
Cloning git@github.com:poga/spacer.git into services/poga/spacer ...
Building services/poga/spacer/examples/hello/docker-compose.yml ...
...
...
Successfully built 2f880daf95fc

Starting services/poga/spacer/examples/hello/docker-compose.yml ...
Proxying /counter to http://0.0.0.0:5000
Proxying /hello to http://0.0.0.0:32770

Spacer is ready and rocking at 0.0.0.0:9064

$ curl 0.0.0.0:9064/counter
{
  "result": "41"
}

$ curl 0.0.0.0:9064/hello
Hello World
```

## Development

1. git clone `git clone git@github.com:poga/spacer.git`
2. Setup Spacerfile, each line is a microservice repo on github. we also support sub folder in a repo. For example:
    ```
    poga/spacer/examples/counter
    poga/spacer/examples/hello
    ```
    Checkout https://github.com/poga/spacer/tree/master/examples/counter and https://github.com/poga/spacer/tree/master/examples/hello for example services.

3. `go run *.go`

## Todo

- [x] Download and Boot up
- [ ] Scaling service with docker-compose
- [ ] Monitering service with ?
- [ ] Conquer the world with Docker

## Why

We already have great tools to help us dealing with library dependencies,
such as Bundler, NPM, and Cargo.
However, in the world of microservices, we still have to manage our own microservice infrastructure.
Creating a scalable microservice infrastructure is a hard task and need a lot of experiences.

With Spacer, you simply write down the service you need, the version you want.
Spacer will take care of all the work of building, monitering, and scaling for you.

For example, we need a production-ready open-source spam-filter service. we can write "poga/spam-fighter" in a file named Spacerfile. Spacer will pull the correct service from github and deploy them to development environment or production-ready IaaS.

Now we can scale this spam-filter service together and everyone can benefit from it. Thanks to Docker.

# Note

This project is an entry of Docker Global Hack Day #3


text:
<?php

class news {

    private $id;
    private $date;
    private $title;
    private $content;

    function news($id = 0) {
        if ($id > 0) {
            /*
              $result = get_cache('news','news_'.$id);
              if (is_bool($result))
              {
             */
            $sql = "SELECT * FROM `news` WHERE id = $id";
            $result = loadSqlResultArray($sql);

            foreach ($result as $key => $value) {
                $this->$key = $value;
            }
        }
    }

// Ascenseur

    function setId($id) {
        $this->id = $id;
    }

    function getId() {
        return $this->id;
    }

    function setDate($date) {
        $this->date = $date;
    }

    function getDate2() {
        return $this->date;
    }

    function setTitle($title) {
        $this->title = $title;
    }

    function getTitle() {
        return $this->title;
    }

    function setContent($content) {
        $this->content = $content;
    }

    function getContent() {
        return $this->content;
    }

// Fonctions

    function showNews() {
        echo '
';
        echo ' ' . $this->getDate2();
        echo ' : ';
        echo $this->getTitle() . ' : ';
        echo '
';

        echo '
';
        echo html_entity_decode($this->getContent());
        echo '

';
    }

// Fonctions static

    public static function getAllNews() {
        $result2 = get_cache('news', 'news_lastnews');
        if (is_bool($result2)) {
            $sql = "SELECT id FROM `news` ORDER BY `date` DESC,id DESC LIMIT 5 ";
            $result = loadSqlResultArrayList($sql);
            create_cache('news', 'news_lastnews', $result2);
            return $result;
        }
    }

    /** Créées un objet News.
     *
     * @param type $title
     * @param type $content
     * @throws Exception
     */
    public static function addNews($title,$content)
    {
        try
        {

            $sql = "INSERT INTO totototo (date,title,content) VALUES (CURRENT_TIMESTAMP,'" . $title . "','" . $content . "')";

            loadSqlExecute($sql);
            destroy_cache('news', 'news_lastnews');
        }
        catch(Exception $e)
        {
            throw new Exception("Une erreur est survenue lors de l'ajout de la news");
        }

    }

}

?>


text:
// Copyright 2014 The Oppia Authors. All Rights Reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//      http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS-IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

/**
 * @fileoverview Page object for the exploration player, for use in Protractor
 * tests.
 */

var forms = require('./forms.js');
var waitFor = require('./waitFor.js');
var interactions = require('../../../extensions/interactions/protractor.js');

var ExplorationPlayerPage = function() {
  var conversationInput = element(
    by.css('.protractor-test-conversation-input'));
  var nextCardButton = element(by.css('.protractor-test-next-card-button'));
  var suggestionDescriptionInput = element(
    by.css('.protractor-test-suggestion-description-input'));
  var closeSuggestionModalButton = element(
    by.css('.protractor-test-exploration-close-suggestion-modal-btn'));
  var conversationSkinCardsContainer = element(
    by.css('.protractor-test-conversation-skin-cards-container'));
  var conversationContent = element.all(
    by.css('.protractor-test-conversation-content'));
  var conversationFeedback = element.all(
    by.css('.protractor-test-conversation-feedback'));
  var explorationHeader = element(
    by.css('.protractor-test-exploration-header'));
  var infoCardRating = element(
    by.css('.protractor-test-info-card-rating'));
  var explorationSuggestionModal = element(
    by.css('.protractor-test-exploration-suggestion-modal'));
  var feedbackTextArea = element(
    by.css('.protractor-test-exploration-feedback-textarea'));
  var waitingForResponseElem = element(by.css(
    '.protractor-test-input-response-loading-dots'));
  var ratingStars = element.all(by.css('.protractor-test-rating-star'));
  var suggestionSubmitButton = element(
    by.css('.protractor-test-suggestion-submit-btn'));
  var feedbackCloseButton = element(
    by.css('.protractor-test-exploration-feedback-close-button'));
  var reportExplorationButton = element(
    by.css('.protractor-test-report-exploration-button'));
  var feedbackSubmitButton = element(
    by.css('.protractor-test-exploration-feedback-submit-btn'));
  var explorationInfoIcon = element(
    by.css('.protractor-test-exploration-info-icon'));
  var nextCardButton = element(
    by.css('.protractor-test-continue-to-next-card-button'));
  var viewHintButton = element(by.css('.protractor-test-view-hint'));
  var viewSolutionButton = element(by.css('.protractor-test-view-solution'));
  var continueToSolutionButton = element(
    by.css('.protractor-test-continue-to-solution-btn'));
  var gotItButton = element(by.css('.oppia-learner-got-it-button'));
  var confirmRedirectionButton =
      element(by.css('.protractor-test-confirm-redirection-button'));
  var cancelRedirectionButton = element(
    by.css('.protractor-test-cancel-redirection-button'));
  var returnToParentButton = element(
    by.css('.protractor-test-return-to-parent-button'));
  var correctFeedbackElement = element(
    by.css('.protractor-test-correct-feedback'));

  var feedbackPopupLink =
    element(by.css('.protractor-test-exploration-feedback-popup-link'));
  var suggestionPopupLink =
    element(by.css('.protractor-test-exploration-suggestion-popup-link'));

  var audioBarExpandButton = element(by.css('.protractor-test-audio-bar'));
  var voiceoverLanguageSelector = element(
    by.css('.protractor-test-audio-lang-select'));
  var playButton = element(by.css('.protractor-test-play-circle'));
  var pauseButton = element(by.css('.protractor-test-pause-circle'));

  this.expandAudioBar = async function() {
    await waitFor.elementToBeClickable(
      audioBarExpandButton, 'Audio bar taking too long to be clickable');
    await audioBarExpandButton.click();
  };

  this.pressPlayButton = async function() {
    await waitFor.elementToBeClickable(
      playButton, 'Play button taking too long to be clickable');
    await playButton.click();
  };

  this.expectAudioToBePlaying = async function() {
    expect(await pauseButton.isPresent()).toBeTruthy();
  };

  this.pressPauseButton = async function() {
    await waitFor.elementToBeClickable(
      pauseButton, 'Pause button taking too long to be clickable');
    await pauseButton.click();
  };

  this.expectAudioToBePaused = async function() {
    expect(await playButton.isPresent()).toBeTruthy();
  };

  this.changeVoiceoverLanguage = async function(language) {
    await waitFor.visibilityOf(
      voiceoverLanguageSelector, 'Language selector takes too long to appear.');
    await voiceoverLanguageSelector.element(
      by.cssContainingText('option', language)).click();
  };

  this.clickThroughToNextCard = async function() {
    await waitFor.elementToBeClickable(
      nextCardButton, '"Next Card" button takes too long to be clickable');
    await nextCardButton.click();
  };

  this.clickSuggestChangesButton = async function() {
    await waitFor.elementToBeClickable(
      suggestionPopupLink, 'Suggest changes button taking too long to appear');
    await suggestionPopupLink.click();
  };

  this.expectNextCardButtonTextToBe = async function(text) {
    await waitFor.visibilityOf(
      nextCardButton, 'Next Card Button not showing up.');
    var buttonText = await nextCardButton.getText();
    expect(buttonText).toMatch(text);
  };

  this.fillAndSubmitSuggestion = async function(
      suggestionTitle, suggestionDescription) {
    var suggestionModal = element(
      by.css('.protractor-test-exploration-suggestion-modal'));
    await waitFor.visibilityOf(
      suggestionModal, 'Suggestion Modal is taking too long to appear.');
    var suggestionHeader = element(by.css('.oppia-rte'));
    await suggestionHeader.click();
    await suggestionHeader.sendKeys(suggestionTitle);
    var suggestionModalDescription = element(
      by.css('.protractor-test-suggestion-description-input'));
    await suggestionModalDescription.click();
    await suggestionModalDescription.sendKeys(suggestionDescription);
    var submitSuggestionBtn = element(
      by.css('.protractor-test-suggestion-submit-btn'));

    await submitSuggestionBtn.click();
    var AFTER_SUBMIT_RESPONSE_STRING =
        'Your suggestion has been forwarded to the ' +
        'exploration author for review.';
    var afterSubmitModalText = await element(by.tagName('p')).getText();
    expect(afterSubmitModalText).toMatch(AFTER_SUBMIT_RESPONSE_STRING);
  };

  this.reportExploration = async function() {
    await waitFor.elementToBeClickable(
      reportExplorationButton,
      'Report Exploration Button takes too long to be clickable');
    await reportExplorationButton.click();
    let radioButton = await element.all(by.tagName('input')).get(0);
    await waitFor.visibilityOf(
      radioButton, 'Radio Buttons takes too long to appear');
    await radioButton.click();
    let textArea = element(by.tagName('textarea'));
    await textArea.sendKeys('Reporting this exploration');
    let submitButton = await element.all(by.tagName('button')).get(1);
    await submitButton.click();
    let afterSubmitText = await element(by.tagName('p')).getText();
    expect(afterSubmitText).toMatch(
      'Your report has been forwarded to the moderators for review.');
  };

  this.viewHint = async function() {
    // We need to wait some time for the solution to activate.
    await waitFor.elementToBeClickable(
      viewHintButton, '"View Hint" button takes too long to be clickable');
    await viewHintButton.click();
    await clickGotItButton();
  };

  this.viewSolution = async function() {
    // We need to wait some time for the solution to activate.
    await waitFor.elementToBeClickable(
      viewSolutionButton,
      '"View Solution" button takes too long to be clickable');
    await viewSolutionButton.click();
    await waitFor.elementToBeClickable(
      continueToSolutionButton,
      '"Continue Solution" button takes too long to be clickable');
    await continueToSolutionButton.click();
    await clickGotItButton();
  };

  var clickGotItButton = async function() {
    await waitFor.elementToBeClickable(
      gotItButton, '"Got It" button takes too long to be clickable');
    await gotItButton.click();
  };

  this.clickConfirmRedirectionButton = async function() {
    await waitFor.elementToBeClickable(
      confirmRedirectionButton,
      '"Confirm Redirect" button takes too long to be clickable');
    await confirmRedirectionButton.click();
    await waitFor.pageToFullyLoad();
  };

  this.clickCancelRedirectionButton = async function() {
    await waitFor.elementToBeClickable(
      cancelRedirectionButton,
      '"Cancel Redirect" button takes too long to be clickable');
    await cancelRedirectionButton.click();
  };

  this.clickOnReturnToParentButton = async function() {
    await waitFor.elementToBeClickable(
      returnToParentButton,
      '"Return to Parent" button takes too long to be clickable');
    await returnToParentButton.click();
    await waitFor.pageToFullyLoad();
  };

  this.clickOnCloseSuggestionModalButton = async function() {
    await waitFor.elementToBeClickable(
      closeSuggestionModalButton,
      '"Close suggestion modal" button takes too long to be clickable');
    await closeSuggestionModalButton.click();
    await waitFor.pageToFullyLoad();
  };

  // This verifies the question just asked, including formatting and
  // rich-text components. To do so the richTextInstructions function will be
  // sent a handler (as given in forms.RichTextChecker) to which calls such as
  //   handler.readItalicText('slanted');
  // can then be sent.
  this.expectContentToMatch = async function(richTextInstructions) {
    await waitFor.visibilityOf(
      await conversationContent.first(), 'Conversation not visible');
    await forms.expectRichText(
      await conversationContent.last()
    ).toMatch(richTextInstructions);
  };

  this.expectExplorationToBeOver = async function() {
    await waitFor.visibilityOf(
      conversationContent.last(), 'Ending message not visible');
    expect(
      await (conversationContent.last()).getText()
    ).toEqual('Congratulations, you have finished!');
  };

  this.expectExplorationToNotBeOver = async function() {
    await waitFor.visibilityOf(
      conversationContent.last(), 'Ending message not visible');
    expect(
      await (conversationContent.last()).getText()
    ).not.toEqual('Congratulations, you have finished!');
  };

  // Additional arguments may be sent to this function, and they will be
  // passed on to the relevant interaction's detail checker.
  this.expectInteractionToMatch = async function(interactionId) {
    await waitFor.visibilityOf(
      conversationSkinCardsContainer,
      'Conversation skill cards take too long to appear.');
    // Convert additional arguments to an array to send on.
    var args = [conversationInput];
    for (var i = 1; i < arguments.length; i++) {
      args.push(arguments[i]);
    }
    var interaction = await interactions.getInteraction(interactionId);
    await interaction.expectInteractionDetailsToMatch.apply(null, args);
  };

  // Note that the 'latest' feedback may be on either the current or a
  // previous card.
  this.expectLatestFeedbackToMatch = async function(richTextInstructions) {
    await forms.expectRichText(
      await conversationFeedback.last()
    ).toMatch(richTextInstructions);
  };

  this.expectExplorationNameToBe = async function(name) {
    expect(
      await explorationHeader.getText()
    ).toBe(name);
  };

  this.expectExplorationRatingOnInformationCardToEqual = async function(
      ratingValue) {
    await explorationInfoIcon.click();
    var value = await infoCardRating.getText();
    expect(value).toBe(ratingValue);
  };

  this.rateExploration = async function(ratingValue) {
    var elements = ratingStars;
    await waitFor.elementToBeClickable(
      await elements.get(ratingValue - 1),
      'Rating Star takes too long to be clickable');
    await (await elements.get(ratingValue - 1)).click();
    await waitFor.elementToBeClickable(
      feedbackCloseButton, 'Close Feedback button is not clickable');
    await feedbackCloseButton.click();
    await waitFor.invisibilityOf(
      feedbackCloseButton, 'Close Feedback button does not disappear');
  };

  // `answerData` is a variable that is passed to the
  // corresponding interaction's protractor utilities.
  // Its definition and type are interaction-specific.
  this.submitAnswer = async function(interactionId, answerData) {
    // The .first() targets the inline interaction, if it exists. Otherwise,
    // it will get the supplemental interaction.
    await interactions.getInteraction(interactionId).submitAnswer(
      conversationInput, answerData);
    await waitFor.invisibilityOf(
      waitingForResponseElem, 'Response takes too long to appear');
  };

  this.submitFeedback = async function(feedback) {
    await waitFor.elementToBeClickable(
      feedbackPopupLink, 'Feedback Popup link takes too long to be clickable');
    await feedbackPopupLink.click();
    await feedbackTextArea.sendKeys(feedback);
    await waitFor.elementToBeClickable(
      feedbackSubmitButton,
      'Feedback Submit button takes too long to be clickable');
    await feedbackSubmitButton.click();
    await waitFor.invisibilityOf(
      feedbackSubmitButton, 'Feedback popup takes too long to disappear');
  };

  this.submitSuggestion = async function(suggestion, description) {
    await waitFor.elementToBeClickable(
      suggestionPopupLink, 'Suggestion Popup link takes too long to appear');
    await suggestionPopupLink.click();
    var editor = await forms.RichTextEditor(explorationSuggestionModal);
    await editor.setPlainText(suggestion);
    await suggestionDescriptionInput.sendKeys(description);
    await waitFor.elementToBeClickable(
      suggestionSubmitButton,
      'Suggestion Submit button takes too long to be clickable');
    await suggestionSubmitButton.click();
    await waitFor.invisibilityOf(
      suggestionSubmitButton, 'Suggestion popup takes too long to disappear');
  };

  this.expectCorrectFeedback = async function() {
    await waitFor.visibilityOf(
      correctFeedbackElement,
      'Correct feedback footer takes too long to appear');
  };
};

exports.ExplorationPlayerPage = ExplorationPlayerPage;