public interface Decoder<U, T> {
    /**
     * Decode the specified object of type U into object of type T.
     *
     * @param e an {@link Event} type. This can be used to differentiate event received.
     * @param s a object of type U
     * @return a new object of type T
     */
    T decode(Event e, U s);

    /**
     * A Decoder may return an instance of a Decoded object to prevent some messages from being delivered to a {@link Function},
     * by returning a {@link Decoded#ABORT} instance. Returning an instance of Decoded with the action set to {@link Decoded.ACTION#CONTINUE}
     * will allow dispatching messages to {@link Function}
     * @param <T> destination type
     */
    public final static class Decoded<T> {
        /**
         * Use this object to prevent the delivering of messages to {@link Function}
         */
        public final static Decoded ABORT = new Decoded(null, ACTION.ABORT);

        public enum ACTION {
            /**
             * Continue the decoding/dispatching process.
             */
            CONTINUE,
            /**
             * Do not dispatch the message to {@link Function}
             */
            ABORT}

        private final T decodedMessage;
        private final ACTION action;

        /**
         * Create a decoded object
         * @param decodedMessage the decoded message
         * @param action the action
         */
        public Decoded(T decodedMessage, ACTION action) {
            this.decodedMessage = decodedMessage;
            this.action = action;
        }

        /**
         * Create a decoded object with action set to {@link ACTION#CONTINUE}
         * @param decodedMessage the decoded Message
         */
        public Decoded(T decodedMessage) {
            this.decodedMessage = decodedMessage;
            this.action = ACTION.CONTINUE;
        }

        /**
         * Return the decoded message.
         * @return the decoded message.
         */
        public T decoded() {
            return decodedMessage;
        }

        /**
         * Return the action.
         * @return the action.
         */
        public ACTION action() {
            return action;
        }
    }

} public enum Event {

    /**
     * This event is fired when the connection is open. This event is only fired once.
     */
    OPEN,

    /**
     * This event is fired when the connection gets closed. This event is only fired once.
     */
    CLOSE,

    /**
     * This event is fired every time a new message is received.
     */
    MESSAGE,

    /**
     * This event is fired every time the connection re-opened
     */
    REOPENED,

    /**
     * This event is fired when unexpected error happens. This event is only fired once.
     */
    ERROR,

    /**
     * This event is fire when the response's status is received. This event is only fired once.
     */
    STATUS,

    /**
     * This event is fire when the response's headers is received. This event is only fired once.
     */
    HEADERS,

    /**
     * This event is fire when the response's bytes are received.
     */
    MESSAGE_BYTES,

    /**
     * This event is fire when the connection has been established. The Transport propagated is the one that worked.
     * This event is only fired once.
     */
    TRANSPORT

} public interface FunctionResolver {

    FunctionResolver DEFAULT = new FunctionResolver() {
        /**
         * {@inheritDoc}
         */
        @Override
        public boolean resolve(String message, Object functionName, FunctionWrapper fn) {
            return (fn.functionName().length() == 0 && !find(message)) || functionName.toString().equalsIgnoreCase(fn.functionName());
        }

        /**
         * Discard empty function name fired by the library
         * @param functionName
         * @return
         */
        private boolean find(String functionName) {
            for (Event e: Event.values()) {
                if (e.name().equalsIgnoreCase(functionName)) {
                    return true;
                }
            }
            return false;
        }
    };

    /**
     * Resolve the current message
     * @param message the original response's body received
     * @param functionName the default function name taken from {@link org.atmosphere.wasync.Event} or from a custom FunctionResolver
     * @param fn The current {@link FunctionWrapper}
     * @return true if the {@link Function} can be invoked.
     */
    boolean resolve(String message, Object functionName, FunctionWrapper fn);
} public class FunctionWrapper {

    private final String functionName;
    private final Function<?> function;

    public FunctionWrapper(String functionName, Function<?> function) {
        this.functionName = functionName;
        this.function = function;
    }

    public Function<?> function(){
        return function;
    }

    public String functionName() {
        return functionName;
    }
} public interface Future extends java.util.concurrent.Future<Socket> {
    /**
     * Send data to the remote Server.
     * @param message the message to fire
     * @return a {@link Future}
     * @throws java.io.IOException
     */
    Future fire(Object message) throws IOException;
    /**
     * Mark the future done. If an exception occurred, this method will throw it.
     * @return a {@link Future}
     */
    Future finishOrThrowException() throws IOException;

    /**
     * If an exception occurs, the {@link Transport} will set it using this method. An application can also
     * use that method to interrupt a blocking {@link Socket#open(Request)} operation. This operation
     * must unlock the current blocked thread.
     * @param t a {@link IOException}
     * @return a {@link Future}
     */
    Future ioException(IOException t);

    /**
     * Mark this instance as done.
     */
    void done();

    /**
     * Close the underlying {@link Socket}
     */
    void close();
} public interface Options {
    /**
     * The used {@link Transport}
     *
     * @return {@link Transport}
     */
    public Transport transport();

    /**
     * Reconnect after a network failure or when the server close the connection.
     *
     * @return reconnect
     */
    public boolean reconnect();

    /**
     * The delay, in milliseconds, before reconnecting.
     *
     * @return The delay, in milliseconds, before reconnecting.
     */
    public int reconnectTimeoutInMilliseconds();

    /**
     * Maximum reconnection attempts that will run in the interval defined by {@link #reconnectTimeoutInMilliseconds}
     *
     * @return the number of maximum reconnection attempts
     */
    public int reconnectAttempts();

    /**
     * When using long-polling and the {@link Request}, the delay before considering the long-polling connection has been fully processed by the server. If you use
     * the {@link org.atmosphere.wasync.impl.AtmosphereClient}, the server will send some handshake so this value is not needed.
     *
     * @return the delay before considering the long-polling connection has been fully processed
     */
    public long waitBeforeUnlocking();

    /**
     * The {@link AsyncHttpClient} used to communicate with server.
     *
     * @return {@link AsyncHttpClient} used to communicate with server.
     */
    public AsyncHttpClient runtime();

    /**
     * Set the {@link AsyncHttpClient}.
     *
     * @param client the {@link AsyncHttpClient}
     */
    public void runtime(AsyncHttpClient client);

    /**
     * Return true is the {@link AsyncHttpClient} is shared between {@link Socket}. Default is false. You need to invoke {@link #runtime(com.ning.http.client.AsyncHttpClient)} to make
     * it shared.
     *
     * @return true is the {@link AsyncHttpClient}
     */
    public boolean runtimeShared();

    /**
     * The time, in seconds to wait before closing the connection.
     *
     * @return The time, in seconds to wait before closing the connection.
     */
    public int requestTimeoutInSeconds();

    /**
     * Return true if binary data, instead of String/Text message, are sent back by the server. Default is false.
     *
     * @return true if binary data,
     */
    boolean binary();

} public interface Request {

    public enum METHOD {GET, POST, TRACE, PUT, DELETE, OPTIONS}
    public enum TRANSPORT {WEBSOCKET, SSE, STREAMING, LONG_POLLING}

    /**
     * The list of transports to try
     * @return a {@link TRANSPORT}
     */
    List<TRANSPORT> transport();

    /**
     * The method
     * @return a {@link METHOD}
     */
    METHOD method();

    /**
     * Return the list of headers
     * @return a Map of headers
     */
    HttpHeaders headers();

    /**
     * Return the list of query params
     * @return a Map of headers
     */
    Map<String, List<String>> queryString();

    /**
     * The list of {@link Encoder} to use before the request is sent.
     * @return The list of {@link Encoder}
     */
    List<Encoder<?,?>> encoders();

    /**
     * The list of {@link Decoder} to use before the request is sent.
     * @return The list of {@link Decoder}
     */
    List<Decoder<?,?>> decoders();

    /**
     * The targetted URI
     * @return the targetted URI
     */
    String uri();

    /**
     * The {@link FunctionResolver} associated with that request.
     * @return The {@link FunctionResolver}
     */
    FunctionResolver functionResolver();

} public interface Socket {

    /**
     * The current state of the underlying Socket.
     */
    public enum STATUS {
        /**
         * The socket is not yet connected
         */
        INIT,
        /**
         * The socket is open and ready to send messages
         */
        OPEN,
        /**
         * The socket was closed and re-opened
         */
        REOPENED,
        /**
         * The socket is close
         */
        CLOSE,
        /**
         * The socket is broken
         */
        ERROR }

    /**
     * Send data to the remote Server. The object will first be delivered to the set of {@link Encoder}, and then send to the server.
     * The server's response will be delivered to the set of defined {@link Function} using the opened {@link Transport}, e.g for
     * {@link Request.TRANSPORT#WEBSOCKET}, the same connection will be re-used and, for others transports, the suspended connection.
     * @param data object to send
     * @return a {@link Future}
     * @throws IOException
     */
    Future fire(Object data) throws IOException;

    /**
     * Associate a {@link Function} with the Socket. When a response is received, the library will try to associated
     * the decoded message (decoded by {@link Decoder}) to the defined type of the {@link Function}
     * @param function a {@link Function}
     * @return this
     */
    Socket on(Function<?> function);

    /**
     * Associate a {@link Function} with the Socket. When a response is received, the library will try to associated
     * the decoded message (decoded by {@link Decoder}) to the defined type of the {@link Function}. The default messages
     * are defined by {@link org.atmosphere.wasync.Event} but handling of custom message can be done using a {@link FunctionResolver}
     * @param function a {@link Function}
     * @return this
     */
    Socket on(String functionMessage, Function<?> function);

    /**
     * Associate a {@link Function} with an {@link Event}. When the event happen the library will try to associated
     * the decoded event (decoded by {@link Decoder}) to the defined type of the {@link Function}. The default event
     * are defined by {@link org.atmosphere.wasync.Event} but handling of custom event can be done using a {@link FunctionResolver}
     * @param function a {@link Function}
     * @return this
     */
    Socket on(Event event, Function<?> function);

    /**
     * Connect to the remote Server using the {@link Request}'s information.
     * @param request a {@link Request}
     * @return this
     * @throws IOException in case the connect fails or a network failure occurs.
     */
    Socket open(Request request) throws IOException;
    
    /**
     * Connect to the remote Server using the {@link Request}'s information, will timeout if the connection failed to open 
     * within a certain time.
     * @param request a {@link Request}
     * @param timeout the maximum time to wait
     * @param unit the time unit of the timeout argument
     * @return this
     * @throws IOException in case the connect fails or a network failure occurs.
     */
    Socket open(Request request, long timeout, TimeUnit unit) throws IOException;

    /**
     * Close this Socket, asynchronously.
     */
    void close();

    /**
     *  Return the {@link STATUS} of this Socket.
     */
    STATUS status();
} public enum STATUS {
    /**
     * The socket is not yet connected
     */
    INIT,
    /**
     * The socket is open and ready to send messages
     */
    OPEN,
    /**
     * The socket was closed and re-opened
     */
    REOPENED,
    /**
     * The socket is close
     */
    CLOSE,
    /**
     * The socket is broken
     */
    ERROR } public interface Transport {

    /**
     * The transport name
     *
     * @return transport name
     */
    Request.TRANSPORT name();

    /**
     * Register a new {@link FunctionResolver}
     *
     * @param function {@link FunctionResolver}
     * @return this;
     */
    Transport registerF(FunctionWrapper function);

    /**
     * Called when an unexpected exception ocurred.
     *
     * @param t a {@link Throwable}
     */
    void onThrowable(Throwable t);

    /**
     * Close the underlying transport}
     */
    void close();

    /**
     * Return the current {@link org.atmosphere.wasync.Socket.STATUS}
     */
    Socket.STATUS status();

    /**
     * Error handled by a {@link Function}
     */
    boolean errorHandled();

    /**
     * Set a {@link Throwable}
     *
     * @param e {@link Throwable}
     */
    void error(Throwable e);

    /**
     * Set the {@link Future}, which can be used to clone the connection.
     *
     * @param f {@link Future}
     */
    void future(Future f);

    /**
     * Set the {@link Future}, which will unlock the {@link Socket#fire} method once the connection has been fully established.
     * @param f {@link Future}
     */
    void connectedFuture(Future f);
} public class Utils {

    public final static boolean whiteSpace(byte[] b) {
        int i = b.length;
        while (i-- > 0 && (b[i] == 10 || b[i] == 32)) {
        }
        return i == -1;
    }
}
import io.netty.handler.codec.http.HttpHeaders;
import org.asynchttpclient.HttpResponseBodyPart;
import org.asynchttpclient.HttpResponseStatus;
import org.asynchttpclient.ListenableFuture;
import org.asynchttpclient.RequestBuilder;
import org.asynchttpclient.netty.ws.NettyWebSocket;
import org.asynchttpclient.ws.WebSocket;
import org.asynchttpclient.ws.WebSocketListener;
import org.asynchttpclient.ws.WebSocketUpgradeHandler;
import org.atmosphere.wasync.Decoder;
import org.atmosphere.wasync.Event;
import org.atmosphere.wasync.FunctionResolver;
import org.atmosphere.wasync.FunctionWrapper;
import org.atmosphere.wasync.Future;
import org.atmosphere.wasync.Options;
import org.atmosphere.wasync.Request;
import org.atmosphere.wasync.Socket;
import org.atmosphere.wasync.Socket.STATUS;
import org.atmosphere.wasync.Transport;
import org.atmosphere.wasync.util.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import static org.atmosphere.wasync.Event.CLOSE;
import static org.atmosphere.wasync.Event.ERROR;
import static org.atmosphere.wasync.Event.HEADERS;
import static org.atmosphere.wasync.Event.MESSAGE;
import static org.atmosphere.wasync.Event.OPEN;
import static org.atmosphere.wasync.Event.REOPENED;
import static org.atmosphere.wasync.Event.STATUS;
import static org.atmosphere.wasync.Event.TRANSPORT;
/*
 * Copyright 2008-2022 Async-IO.org
 *
 * 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.
 */
package org.atmosphere.wasync.transport;




/**
 * WebSocket {@link org.atmosphere.wasync.Transport} implementation
 *
 * @author Jeanfrancois Arcand
 */
public class WebSocketTransport extends WebSocketUpgradeHandler implements Transport {

	private final Logger logger = LoggerFactory.getLogger(WebSocketTransport.class);
	private NettyWebSocket webSocket;

	private final AtomicBoolean ok = new AtomicBoolean(false);
	private final AtomicInteger reconnectAttempt = new AtomicInteger();
	private final AtomicBoolean reconnecting = new AtomicBoolean(false);

	private final List<FunctionWrapper> functions;
	private final List<Decoder<?, ?>> decoders;
	private final FunctionResolver resolver;
	private final Options options;
	private final RequestBuilder requestBuilder;
	private final AtomicBoolean closed = new AtomicBoolean(false);
	private STATUS status = Socket.STATUS.INIT;
	private final AtomicBoolean errorHandled = new AtomicBoolean();
	private Future underlyingFuture;