/**
 * Copyright 2007-2016, Kaazing Corporation. 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.
 */
package org.kaazing.gateway.transport.wsn;

import static java.lang.String.format;
import static java.util.Arrays.asList;
import static org.kaazing.gateway.resource.address.ResourceAddress.BIND_ALTERNATE;
import static org.kaazing.gateway.resource.address.ResourceAddress.NEXT_PROTOCOL;
import static org.kaazing.gateway.resource.address.ResourceAddress.TRANSPORT;
import static org.kaazing.gateway.resource.address.URLUtils.appendURI;
import static org.kaazing.gateway.resource.address.URLUtils.ensureTrailingSlash;
import static org.kaazing.gateway.resource.address.http.HttpResourceAddress.REALMS;
import static org.kaazing.gateway.resource.address.ws.WsResourceAddress.CODEC_REQUIRED;
import static org.kaazing.gateway.resource.address.ws.WsResourceAddress.INACTIVITY_TIMEOUT;
import static org.kaazing.gateway.resource.address.ws.WsResourceAddress.LIGHTWEIGHT;
import static org.kaazing.gateway.resource.address.ws.WsResourceAddress.MAX_MESSAGE_SIZE;
import static org.kaazing.gateway.transport.http.HttpAcceptor.BALANCEES_KEY;
import static org.kaazing.gateway.transport.http.bridge.filter.HttpMergeRequestFilter.DRAFT76_KEY3_BUFFER_KEY;
import static org.kaazing.gateway.transport.http.bridge.filter.HttpSubjectSecurityFilter.AUTH_SCHEME_APPLICATION_PREFIX;
import static org.kaazing.gateway.transport.ws.util.WsUtils.ACTIVE_EXTENSIONS_KEY;
import static org.kaazing.gateway.transport.ws.util.WsUtils.HEADER_WEBSOCKET_EXTENSIONS;
import static org.kaazing.gateway.transport.ws.util.WsUtils.HEADER_X_WEBSOCKET_EXTENSIONS;
import static org.kaazing.gateway.transport.ws.util.WsUtils.negotiateWebSocketProtocol;
import static org.kaazing.gateway.transport.wsn.WsnSession.SESSION_KEY;
import static org.kaazing.gateway.util.ws.WebSocketWireProtocol.HYBI_13;
import static org.kaazing.mina.core.buffer.IoBufferEx.FLAG_NONE;

import java.io.IOException;
import java.net.ProtocolException;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.ByteBuffer;
import java.nio.charset.CharacterCodingException;
import java.nio.charset.Charset;
import java.util.Collection;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.Callable;
import java.util.concurrent.ScheduledExecutorService;

import javax.annotation.Resource;
import javax.security.auth.Subject;

import org.apache.mina.core.filterchain.IoFilter;
import org.apache.mina.core.filterchain.IoFilterChain;
import org.apache.mina.core.future.CloseFuture;
import org.apache.mina.core.future.IoFuture;
import org.apache.mina.core.future.IoFutureListener;
import org.apache.mina.core.future.WriteFuture;
import org.apache.mina.core.service.IoHandler;
import org.apache.mina.core.service.TransportMetadata;
import org.apache.mina.core.session.AttributeKey;
import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.core.session.IoSessionInitializer;
import org.apache.mina.core.write.WriteRequest;
import org.kaazing.gateway.resource.address.Protocol;
import org.kaazing.gateway.resource.address.ResourceAddress;
import org.kaazing.gateway.resource.address.ResourceAddressFactory;
import org.kaazing.gateway.resource.address.ResourceOptions;
import org.kaazing.gateway.resource.address.URLUtils;
import org.kaazing.gateway.resource.address.http.HttpRealmInfo;
import org.kaazing.gateway.resource.address.uri.URIUtils;
import org.kaazing.gateway.resource.address.ws.WsResourceAddress;
import org.kaazing.gateway.resource.address.wsn.WsnResourceAddressFactorySpi;
import org.kaazing.gateway.security.auth.context.ResultAwareLoginContext;
import org.kaazing.gateway.transport.AbstractBridgeAcceptor;
import org.kaazing.gateway.transport.AbstractBridgeSession;
import org.kaazing.gateway.transport.Bindings;
import org.kaazing.gateway.transport.Bindings.Binding;
import org.kaazing.gateway.transport.BridgeAcceptor;
import org.kaazing.gateway.transport.BridgeServiceFactory;
import org.kaazing.gateway.transport.BridgeSession;
import org.kaazing.gateway.transport.BridgeSessionInitializer;
import org.kaazing.gateway.transport.BridgeSessionInitializerAdapter;
import org.kaazing.gateway.transport.DefaultIoSessionConfigEx;
import org.kaazing.gateway.transport.DefaultTransportMetadata;
import org.kaazing.gateway.transport.IoFilterAdapter;
import org.kaazing.gateway.transport.IoHandlerAdapter;
import org.kaazing.gateway.transport.LoggingUtils;
import org.kaazing.gateway.transport.NioBindException;
import org.kaazing.gateway.transport.TypedAttributeKey;
import org.kaazing.gateway.transport.UpgradeFuture;
import org.kaazing.gateway.transport.http.HttpAcceptSession;
import org.kaazing.gateway.transport.http.HttpAcceptor;
import org.kaazing.gateway.transport.http.HttpProtocol;
import org.kaazing.gateway.transport.http.HttpStatus;
import org.kaazing.gateway.transport.http.HttpUtils;
import org.kaazing.gateway.transport.http.bridge.HttpResponseMessage;
import org.kaazing.gateway.transport.http.bridge.filter.HttpMergeRequestFilter;
import org.kaazing.gateway.transport.http.bridge.filter.HttpProtocolCompatibilityFilter;
import org.kaazing.gateway.transport.ws.WsAcceptor;
import org.kaazing.gateway.transport.ws.WsBinaryMessage;
import org.kaazing.gateway.transport.ws.WsCloseMessage;
import org.kaazing.gateway.transport.ws.WsContinuationMessage;
import org.kaazing.gateway.transport.ws.WsMessage;
import org.kaazing.gateway.transport.ws.WsPingMessage;
import org.kaazing.gateway.transport.ws.WsPongMessage;
import org.kaazing.gateway.transport.ws.WsTextMessage;
import org.kaazing.gateway.transport.ws.bridge.filter.WsBuffer;
import org.kaazing.gateway.transport.ws.bridge.filter.WsBufferAllocator;
import org.kaazing.gateway.transport.ws.bridge.filter.WsCheckAliveFilter;
import org.kaazing.gateway.transport.ws.bridge.filter.WsCodecFilter;
import org.kaazing.gateway.transport.ws.bridge.filter.WsDraftHixieBufferAllocator;
import org.kaazing.gateway.transport.ws.bridge.filter.WsDraftHixieFrameCodecFilter;
import org.kaazing.gateway.transport.ws.bridge.filter.WsFrameBase64Filter;
import org.kaazing.gateway.transport.ws.bridge.filter.WsFrameEncodingSupport;
import org.kaazing.gateway.transport.ws.bridge.filter.WsFrameTextFilter;
import org.kaazing.gateway.transport.ws.bridge.filter.WsFrameUtf8Filter;
import org.kaazing.gateway.transport.ws.extension.ExtensionHelper;
import org.kaazing.gateway.transport.ws.extension.WebSocketExtension;
import org.kaazing.gateway.transport.ws.extension.WebSocketExtensionFactory;
import org.kaazing.gateway.transport.ws.util.WsHandshakeNegotiationException;
import org.kaazing.gateway.transport.ws.util.WsUtils;
import org.kaazing.gateway.util.Encoding;
import org.kaazing.gateway.util.scheduler.SchedulerProvider;
import org.kaazing.gateway.util.ws.WebSocketWireProtocol;
import org.kaazing.mina.core.buffer.IoBufferAllocatorEx;
import org.kaazing.mina.core.buffer.IoBufferEx;
import org.kaazing.mina.core.future.UnbindFuture;
import org.kaazing.mina.core.service.IoProcessorEx;
import org.kaazing.mina.core.session.IoSessionEx;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class WsnAcceptor extends AbstractBridgeAcceptor<WsnSession, WsnBindings.WsnBinding> {

    private static final TypedAttributeKey<ResultAwareLoginContext> LOGIN_CONTEXT_TRANSFER_KEY
                            = new TypedAttributeKey<>(WsnAcceptor.class, "login_transfer");
    private static final TypedAttributeKey<Subject> SUBJECT_TRANSFER_KEY
                            = new TypedAttributeKey<>(WsnAcceptor.class, "subject_transfer");

    static final String CHECK_ALIVE_FILTER = WsnProtocol.NAME + "#checkalive";
	        static final String CODEC_FILTER = WsnProtocol.NAME + "#codec";
	private static final String UTF8_FILTER = WsnProtocol.NAME + "#utf8";
	private static final String BASE64_FILTER = WsnProtocol.NAME + "#base64";
	private static final String TEXT_FILTER = WsnProtocol.NAME + "#text";

    private static final AttributeKey LOCAL_ADDRESS_KEY = new AttributeKey(WsnAcceptor.class, "localAddress");
    private static final AttributeKey REMOTE_ADDRESS_KEY = new AttributeKey(WsnAcceptor.class, "remoteAddress");

    private static final AttributeKey HTTP_REQUEST_URI_KEY = new AttributeKey(WsnAcceptor.class, "httpRequestURI");

    private static final TypedAttributeKey<ResourceAddress> WEBSOCKET_LOCAL_ADDRESS
            = new TypedAttributeKey<>(WsnAcceptor.class, "websocketLocalAddress");

    private static final TypedAttributeKey<String[]> SUPPORTED_PROTOCOLS
            = new TypedAttributeKey<>(WsnAcceptor.class, "supportedProtocols");

    private static final String HEADER_ORIGIN = "Origin";
    private static final String HEADER_CONNECTION = "Connection";
    private static final String HEADER_UPGRADE = "Upgrade";
    private static final String HEADER_WEBSOCKET_ORIGIN = "WebSocket-Origin";
    private static final String HEADER_WEBSOCKET_LOCATION = "WebSocket-Location";
    private static final String HEADER_WEBSOCKET_PROTOCOL = "WebSocket-Protocol";

    private static final String HEADER_X_WEBSOCKET_PROTOCOL = "X-WebSocket-Protocol";

    // "secure" header keys introduced in protocol draft 76
    private static final String HEADER_SEC_WEBSOCKET_LOCATION = "Sec-WebSocket-Location";
    private static final String HEADER_SEC_WEBSOCKET_ORIGIN = "Sec-WebSocket-Origin";
    private static final String HEADER_SEC_WEBSOCKET_PROTOCOL = "Sec-WebSocket-Protocol";
    private static final String HEADER_SEC_WEBSOCKET_EXTENSION = WsUtils.HEADER_SEC_WEBSOCKET_EXTENSIONS;

    private static final String REASON_WEB_SOCKET_HANDSHAKE = "Web Socket Protocol Handshake";
    private static final String HEADER_WEBSOCKET_KEY1 = "Sec-WebSocket-Key1";
    private static final String HEADER_WEBSOCKET_KEY2 = "Sec-WebSocket-Key2";
    private static final String WEB_SOCKET = "WebSocket";
    private static final String WEB_SOCKET_LOWERCASE = "websocket";

    private static final String PROTOCOL_NAME_WEB_SOCKET_RFC6455 = "ws/rfc6455";
    private static final String PROTOCOL_NAME_WEB_SOCKET_DRAFT = "ws/draft-7x";

    // "Sec-WebSocket-Key" used by HyBi Drafts
    private static final String HEADER_WEBSOCKET_KEY = "Sec-WebSocket-Key";
    private static final String HEADER_WEBSOCKET_ACCEPT = "Sec-WebSocket-Accept";
    private static final String HEADER_WEBSOCKET_VERSION = "Sec-WebSocket-Version";
    private static final String WEB_SOCKET_VERSION_KEY = "WebSocketVersion";

    private Properties configuration =  new Properties();
    private ScheduledExecutorService scheduler;
    private BridgeServiceFactory bridgeServiceFactory;
    private ResourceAddressFactory resourceAddressFactory;
    private WebSocketExtensionFactory webSocketExtensionFactory;

    private static final ExtensionHelper extensionHelper = new ExtensionHelper() {

        @Override
        public void setLoginContext(IoSession session, ResultAwareLoginContext loginContext) {
            WsnSession wsnSession = SESSION_KEY.get(session);
            assert wsnSession !=  null;
            wsnSession.setLoginContext(loginContext);
        }

        @Override
        public void closeWebSocketConnection(IoSession session) {
            WsnSession wsnSession = SESSION_KEY.get(session);
            assert wsnSession !=  null;
            wsnSession.close(false);
        }

    };

    public WsnAcceptor() {
        super(new DefaultIoSessionConfigEx());
    }

    @Resource(name = "configuration")
    public void setConfiguration(Properties configuration) {
        this.configuration = configuration;
    }

    @Resource(name = "schedulerProvider")
    public void setSchedulerProvider(SchedulerProvider provider) {
        this.scheduler = provider.getScheduler("Scheduler-wsn", false);
    }

    @Resource(name = "bridgeServiceFactory")
    public void setBridgeServiceFactory(BridgeServiceFactory bridgeServiceFactory) {
        this.bridgeServiceFactory = bridgeServiceFactory;
    }

    @Resource(name = "resourceAddressFactory")
    public void setResourceAddressFactory(ResourceAddressFactory factory) {
        this.resourceAddressFactory = factory;
    }

    @Resource(name = "ws.acceptor")
    public void setWsAcceptor(WsAcceptor acceptor) {
        this.webSocketExtensionFactory = acceptor.getWebSocketExtensionFactory();
    }

    @Override
    protected void init() {
        // Fail gateway startup if the obsolete system property "org.kaazing.gateway.transport.ws.INACTIVITY_TIMEOUT"
        // from JMS Edition release 3.5.3 is used (KG-7125)
        WsCheckAliveFilter.validateSystemProperties(configuration, logger);

        super.init();
    }

    /* for test observalibility only */
    Bindings<WsnBindings.WsnBinding> bindings() {
        return super.bindings;
    }

    @Override
    protected WsnBindings initBindings() {
        return new WsnBindings();
    }

    @Override
    protected IoProcessorEx<WsnSession> initProcessor() {
        return new WsnAcceptProcessor();
    }

    @Override
    protected boolean canBind(String transportName) {
        return transportName.equals("wsn") || transportName.equals("ws");
    }

    @Override
    public void bind(ResourceAddress address,
                     IoHandler handler,
                     BridgeSessionInitializer<? extends IoFuture> initializer) {
        bind0(address, handler, initializer);
    }

    @Deprecated // move to HttpBalancerFilter added to WsnAcceptor.httpBridgeHandler filterChain instead
    private <T extends IoFuture> /*final*/ void bind0(ResourceAddress address, IoHandler handler, final BridgeSessionInitializer<T> initializer) {

        BridgeSessionInitializer<T> balancerInitializer = new BridgeSessionInitializerAdapter<T>() {
            @Override
            public void initializeSession(IoSession session, T future) {
                if (initializer != null) {
                    initializer.initializeSession(session, future);
                }
                final WsnSession wsnSession = (WsnSession) session;
                if (wsnSession.isBalanceSupported()) {
                    // NOTE: this collection is either null, empty or length one.
                    //       the balancee URI is selected in the HttpBalancerService's
                    //       preHttpUpgradeSessionInitializer.  That's why the iterator.next() is assumed here.

                    Collection<String> balanceeURIs = wsnSession.getBalanceeURIs();
                    String response = "" + '\uf0ff'; // Unique prefix to avoid collisions with responses from non Kaazing servers
                                                     // Encoded from UTF-8 will become: 0xef 0x83 0xbf
                    if (balanceeURIs == null) {
                        // No balancer participated in this session initialization
                        response += "N";
                    } else {
                        if (balanceeURIs.isEmpty()) {
                            // Balancer participated in this session initialization but found no balancees
                            response += "R";
                        } else {
                            // Balancer participated in this session initialization and found balancees
                            try {
                                response += "R";
                                response += HttpUtils.mergeQueryParameters(wsnSession.getParentHttpRequestURI(),
                                        balanceeURIs.iterator().next());
                            } catch (URISyntaxException e) {
                                logger.error(
                                        String.format("Failed to manufacture a balancee URI:  The Http Request URI Query '%s' cannot merge with the configured balancee URI '%s'",
                                                wsnSession.getParentHttpRequestURI().getQuery(), balanceeURIs.iterator().next()), e);
                            }
                        }
                    }
                    WebSocketWireProtocol wsVersion = wsnSession.getVersion();

                    WriteFuture writeFuture;

                    Boolean codecRequired = wsnSession.getLocalAddress().getOption(CODEC_REQUIRED);

                    final IoSessionEx parent = wsnSession.getParent();
                    if (WebSocketWireProtocol.HYBI_13.equals(wsVersion) || WebSocketWireProtocol.HYBI_8.equals(wsVersion)) {

                        final IoBufferAllocatorEx<? extends WsBuffer> allocator = wsnSession.getBufferAllocator();
                        ByteBuffer messageNioBuf = allocator.allocate(response.getBytes().length + 4);
                        IoBufferEx messageBuf = allocator.wrap(messageNioBuf).setAutoExpander(allocator);
                        try {
                            messageBuf.putString(response, Charset.forName("UTF-8").newEncoder());
                            messageBuf.flip();
                        } catch (CharacterCodingException e) {
                            logger.error(e.toString());
                        }
                        WsBinaryMessage message = new WsBinaryMessage(messageBuf);

                        if (codecRequired) {
                            IoBufferAllocatorEx<?> parentAllocator = parent.getBufferAllocator();
                            IoBufferEx b = WsFrameEncodingSupport.doEncode(parentAllocator, FLAG_NONE, message);
                            writeFuture = parent.write(b); // Write on parent session to avoid another encoding by the WebSocket session
                        } else {
                            writeFuture = parent.write(message);
                        }
                    } else {
                        IoBufferAllocatorEx<?> parentAllocator = parent.getBufferAllocator();

                        ByteBuffer parentNioBuf = parentAllocator.allocate(response.getBytes().length + 6);
                        IoBufferEx parentBuf = parentAllocator.wrap(parentNioBuf).setAutoExpander(parentAllocator);

                        if ( codecRequired ) { parentBuf.put((byte) 0x00); }
                        try {
                            parentBuf.putString(response, Charset.forName("UTF-8").newEncoder());
                        } catch (CharacterCodingException e) {
                            logger.error(e.toString());
                        }
                        if ( codecRequired ) { parentBuf.put((byte) 0xff); }
                        parentBuf.flip();
                        writeFuture = parent.write(parentBuf); // Write on parent session to avoid encoding on ByteSocket connections
                    }

                }
            }

            @Override
            public BridgeSessionInitializer<T> getParentInitializer(Protocol protocol) {
                return initializer != null ? initializer.getParentInitializer(protocol):super.getParentInitializer(protocol);
            }
        };

        //
        // Bind this address to the handler and initializers.
        //
        super.bind(address, handler, balancerInitializer);
    }


    @Override
    protected <T extends IoFuture> void bindInternal(final ResourceAddress address, IoHandler handler,
                                                     final BridgeSessionInitializer<T> initializer) {
        //
        // Bind the transport
        //
        final BridgeSessionInitializer<T> httpInitializer =
                (initializer != null) ? initializer.getParentInitializer(HttpProtocol.HTTP)
                : null;

        BridgeSessionInitializer<T> wrapperInitializer = new BridgeSessionInitializerAdapter<T>() {
            @Override
            public void initializeSession(IoSession session, T future) {
                if ( httpInitializer != null ) {
                    httpInitializer.initializeSession(session, future);
                }

                String nextProtocol = address.getOption(ResourceAddress.NEXT_PROTOCOL);
                String[] supportedProtocols = address.getOption(WsResourceAddress.SUPPORTED_PROTOCOLS);
                String[] allSupportedProtocols = supportedProtocols;
                if (nextProtocol != null) {
                    // add next-protocol (if specified) to end of supported protocols (whether empty or not)
                    allSupportedProtocols = new String[supportedProtocols.length + 1];
                    System.arraycopy(supportedProtocols, 0, allSupportedProtocols, 0, supportedProtocols.length);
                    allSupportedProtocols[supportedProtocols.length] = nextProtocol;
                }

                // Store the next over-the-top of-websocket protocols the server supports for this address on session.
                SUPPORTED_PROTOCOLS.set(session, allSupportedProtocols);
            }
        };

        bindApiPath(address);

        BridgeAcceptor acceptor = bridgeServiceFactory.newBridgeAcceptor(address.getTransport());
        try {
            ResourceAddress transport = address.getTransport();
            IoHandler transportHandler = selectHandler(address);
            acceptor.bind(transport, transportHandler, wrapperInitializer);
        } catch (NioBindException e) {
            throw new RuntimeException("Unable to bind address " + address.getTransport() + ": " + e.getMessage(), e);
        }
    }

    @Override
    protected UnbindFuture unbindInternal(ResourceAddress address, IoHandler handler,
            BridgeSessionInitializer<? extends IoFuture> initializer) {
        unbindApiPath(address);
        ResourceAddress transport = address.getTransport();
        BridgeAcceptor acceptor = bridgeServiceFactory.newBridgeAcceptor(transport);
        return acceptor.unbind(address.getTransport());
    }

    private void bindApiPath(ResourceAddress address) {
        ResourceAddress apiAddress = createApiAddress(address);
        bridgeServiceFactory.newBridgeAcceptor(apiAddress).bind(apiAddress, WsAcceptor.API_PATH_HANDLER, null);
    }

    private void unbindApiPath(ResourceAddress address) {
        ResourceAddress apiAddress = createApiAddress(address);
        bridgeServiceFactory.newBridgeAcceptor(apiAddress).unbind(apiAddress);
    }

    private ResourceAddress createApiAddress(ResourceAddress address) {
        ResourceAddress transport = address.getTransport();

        ResourceOptions apiAddressOptions = ResourceOptions.FACTORY.newResourceOptions(transport);
        // /;api/operation is a terminal endpoint so next protocol should be null
        apiAddressOptions.setOption(NEXT_PROTOCOL, null);
        // even if the address has an alternate, do not bind the alternate
        apiAddressOptions.setOption(BIND_ALTERNATE, Boolean.FALSE);

        String path = URIUtils.getPath(appendURI(ensureTrailingSlash(address.getExternalURI()), HttpProtocolCompatibilityFilter.API_PATH));
        String apiLocation = URIUtils.modifyURIPath(URIUtils.uriToString(transport.getResource()), path);
        return resourceAddressFactory.newResourceAddress(apiLocation, apiAddressOptions);
    }

    private IoHandler selectHandler(ResourceAddress address) {
        ResourceAddress httpTransport = address.getTransport();
        String httpNextProtocol = httpTransport.getOption(NEXT_PROTOCOL);
        if ( PROTOCOL_NAME_WEB_SOCKET_RFC6455.equals(httpNextProtocol)) {
            return wsnHttpRFC6455BridgeHandler;
        } else if ( PROTOCOL_NAME_WEB_SOCKET_DRAFT.equals(httpNextProtocol)) {
            return wsnHttpDelegatingBridgeHandler;
        }
        return null;
    }


    @Override
    protected IoFuture dispose0() throws Exception {
        scheduler.shutdownNow();
        return super.dispose0();
    }

    @Override
    public TransportMetadata getTransportMetadata() {
        return new DefaultTransportMetadata(WsnProtocol.NAME);
    }

    private IoHandlerAdapter<IoSessionEx> ioBridgeHandler = new IoHandlerAdapter<IoSessionEx>()  {

        private final WsFrameBase64Filter base64 = new WsFrameBase64Filter();
        private final WsFrameUtf8Filter utf8 = new WsFrameUtf8Filter();
        private final WsFrameTextFilter text = new WsFrameTextFilter();

        @Override
        protected void doSessionOpened(final IoSessionEx session) throws Exception {

            IoFilterChain filterChain = session.getFilterChain();

            final ResourceAddress localAddress = WEBSOCKET_LOCAL_ADDRESS.get(session);

            final WebSocketWireProtocol wsVersion = (WebSocketWireProtocol) session.getAttribute(WEB_SOCKET_VERSION_KEY);

            final boolean wasHixieHandshake = wsVersion != null &&
                                              (wsVersion == WebSocketWireProtocol.HIXIE_76 ||
                                               wsVersion == WebSocketWireProtocol.HIXIE_75);

            addBridgeFilters(filterChain, localAddress);

            session.resumeRead();

            //
            // Construct remote address from transport session's remote address.
            //

            ResourceAddress transportAddress = (ResourceAddress) session.getAttribute(REMOTE_ADDRESS_KEY);
            assert transportAddress != null;
            ResourceOptions options = ResourceOptions.FACTORY.newResourceOptions();
            options.setOption(TRANSPORT, transportAddress);
            options.setOption(NEXT_PROTOCOL, localAddress.getOption(NEXT_PROTOCOL));

            // TODO: Verify this
            final ResourceAddress remoteAddress =
                    resourceAddressFactory.newResourceAddress(URIUtils.uriToString(localAddress.getResource()), options);

            //
            // We remember the http uri from the session below us.
            //
            final URI httpUri = (URI) session.getAttribute(HTTP_REQUEST_URI_KEY);

            //
            // Build a new Wsn Session.
            //
            final WsnSession wsnSession = newSession(new IoSessionInitializer<IoFuture>() {
                @Override
                public void initializeSession(IoSession wsnSession, IoFuture future) {
                    wsnSession.setAttribute(HttpAcceptor.SERVICE_REGISTRATION_KEY, session
                              .getAttribute(HttpAcceptor.SERVICE_REGISTRATION_KEY));
                    WsnSession typedWsnSession = (WsnSession) wsnSession;
                    typedWsnSession.setSubject(SUBJECT_TRANSFER_KEY.remove(session));
                    wsnSession.setAttribute(BridgeSession.NEXT_PROTOCOL_KEY, session.getAttribute(BridgeSession.NEXT_PROTOCOL_KEY));
                    ACTIVE_EXTENSIONS_KEY.set(wsnSession, ACTIVE_EXTENSIONS_KEY.get(session));
                    HttpMergeRequestFilter.INITIAL_HTTP_REQUEST_KEY.set(wsnSession, HttpMergeRequestFilter.INITIAL_HTTP_REQUEST_KEY.remove(session));
                    DRAFT76_KEY3_BUFFER_KEY.set(wsnSession, DRAFT76_KEY3_BUFFER_KEY.remove(session));
                    if (HttpEmptyPacketWriterFilter.writeExtraEmptyPacketRequired(wsnSession)) {
                        wsnSession.setAttribute(HttpProtocolCompatibilityFilter.EMPTY_PACKET_PRODUCER_FILTER, HttpEmptyPacketWriterFilter.INSTANCE);
                    }
                }
            }, new Callable<WsnSession>() {
                @Override
                public WsnSession call() {
                    URI httpRequestURI = httpUri;
                    ResultAwareLoginContext loginContext = LOGIN_CONTEXT_TRANSFER_KEY.remove(session);
                    IoBufferAllocatorEx<?> parentAllocator = session.getBufferAllocator();
                    IoBufferAllocatorEx<WsBuffer> allocator = wasHixieHandshake ? new WsDraftHixieBufferAllocator(parentAllocator)
                                                                                : new WsBufferAllocator(parentAllocator, false /* masking */);
                    WsnSession newWsnSession = new WsnSession(WsnAcceptor.this, getProcessor(), localAddress, remoteAddress,
                            session, allocator, httpRequestURI, loginContext,
                            wsVersion, null);

                    IoHandler handler = getHandler(localAddress);

                    newWsnSession.setHandler(handler);
                    newWsnSession.setBridgeServiceFactory(bridgeServiceFactory);
                    newWsnSession.setResourceAddressFactory(resourceAddressFactory);
                    newWsnSession.setScheduler(scheduler);
                    return newWsnSession;
                }
            });

            // Establish session cleanup listener.
            CloseFuture closeFuture = wsnSession.getCloseFuture();
            closeFuture.addListener(new IoFutureListener<CloseFuture>() {
                @Override
                public void operationComplete(CloseFuture future) {
                    if (logger.isTraceEnabled()) {
                        logger.trace(WsnAcceptor.class.getSimpleName()
                                + " removing enforcement of lifetime for closed session (" + wsnSession.getId() + ").");
                    }

                    DRAFT76_KEY3_BUFFER_KEY.remove(wsnSession);

                    wsnSession.shutdownScheduledCommands();
                    wsnSession.logout();
                }
            });

            SESSION_KEY.set(session, wsnSession);

            if ( !session.isClosing() ) {
                wsnSession.startupScheduledCommands();
            }
        }

        @Override
        protected void doMessageReceived(IoSessionEx session, Object message) throws Exception {
            WsnSession wsnSession = SESSION_KEY.get(session);

            if (wsnSession != null) {
                IoFilterChain filterChain = wsnSession.getFilterChain();
                IoBufferAllocatorEx<? extends WsBuffer> allocator = wsnSession.getBufferAllocator();

                final boolean hasPostUpgradeChildWsnSession = wsnSession.getHandler() == ioBridgeHandler;
                final ResourceAddress wsnSessionLocalAddress = wsnSession.getLocalAddress();
                final boolean isLightweightWsnSession = wsnSessionLocalAddress.getOption(LIGHTWEIGHT);
                boolean sendMessagesDirect = isLightweightWsnSession
                                             && hasPostUpgradeChildWsnSession; // post-upgrade
                if ( sendMessagesDirect ) {
                    filterChain.fireMessageReceived(message);
                    return;
                }


                WsMessage wsMessage = (WsMessage) message;
                switch (wsMessage.getKind()) {
                case CONTINUATION:
                    WsContinuationMessage wsCont = (WsContinuationMessage) wsMessage;
                    IoBufferEx wsContBytes = wsCont.getBytes();
                    WsBuffer wsContBuffer = allocator.wrap(wsContBytes.buf());
                    wsContBuffer.setKind(WsBuffer.Kind.CONTINUATION);
                    wsContBuffer.setFin(wsCont.isFin());
                    filterChain.fireMessageReceived(wsContBuffer);
                    break;
                case TEXT:
                    WsTextMessage wsText = (WsTextMessage) wsMessage;
                    IoBufferEx wsTextBytes = wsText.getBytes();
                    WsBuffer wsTextBuffer = allocator.wrap(wsTextBytes.buf());
                    wsTextBuffer.setKind(WsBuffer.Kind.TEXT);
                    wsTextBuffer.setFin(wsText.isFin());
                    filterChain.fireMessageReceived(wsTextBuffer);
                    break;
                case BINARY:
                    WsBinaryMessage wsBinary = (WsBinaryMessage) wsMessage;
                    IoBufferEx wsBinaryBytes = wsBinary.getBytes();
                    WsBuffer wsBinaryBuffer = allocator.wrap(wsBinaryBytes.buf());
                    wsBinaryBuffer.setKind(WsBuffer.Kind.BINARY);
                    wsBinaryBuffer.setFin(wsBinary.isFin());
                    filterChain.fireMessageReceived(wsBinaryBuffer);
                    break;
                case PING:
                    // bounce back PONGs in response to client PINGs
                    WsPingMessage ping = (WsPingMessage) wsMessage;
                    IoBufferEx payload = ping.getBytes();
