Workaround for UNDERTOW-282

io.undertow.servlet.ServletExtension

co.vgw.wildfly.Undertow282Fix

Overview

This solution is based on comment https://issues.jboss.org/browse/UNDERTOW-282?focusedCommentId=12986535&page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel#comment-12986535

This fix was tested with Wildfly 8.1.0 Final.

NOTE: io.undertow.servlet.ServletExtension file should go into META-INF/services/io.undertow.servlet.ServletExtension

pom.xml

            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-war-plugin</artifactId>
                <version>2.4</version>
                <configuration>
                ...
                    <archive>
                        <manifestEntries>
                            <Dependencies>org.jboss.xnio, io.undertow.core</Dependencies>
                        </manifestEntries>
                    </archive>
                ...    
                </configuration>
            </plugin>

Undertow282Fix

package co.vgw.wildfly;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import javax.servlet.ServletContext;

import org.apache.log4j.Logger;
import org.xnio.channels.StreamSinkChannel;
import org.xnio.conduits.AbstractStreamSourceConduit;
import org.xnio.conduits.StreamSourceConduit;

import io.undertow.server.ConduitWrapper;
import io.undertow.server.HandlerWrapper;
import io.undertow.server.HttpHandler;
import io.undertow.server.HttpServerExchange;
import io.undertow.server.ServerConnection;
import io.undertow.servlet.ServletExtension;
import io.undertow.servlet.api.DeploymentInfo;
import io.undertow.util.ConduitFactory;


/**
 * Created by atchijov on 21/7/14.
 */
public class Undertow282Fix implements ServletExtension {
    @Override
    public void handleDeployment(DeploymentInfo deploymentInfo, ServletContext servletContext) {
        deploymentInfo.addInitialHandlerChainWrapper(new HandlerWrapper() {
            @Override
            public HttpHandler wrap(final HttpHandler handler) {
                return new FixSslHandler(handler);
            }
        });
    }

    private static class FixSslHandler implements HttpHandler {
        private final HttpHandler handler;

        public FixSslHandler(HttpHandler handler) {
            this.handler = handler;
        }

        @Override
        public void handleRequest(HttpServerExchange exchange) throws Exception {
            if(!exchange.isRequestComplete()) {
                exchange.addRequestWrapper(new ConduitWrapper<StreamSourceConduit>() {
                    @Override
                    public StreamSourceConduit wrap(ConduitFactory<StreamSourceConduit> factory, HttpServerExchange exchange) {
                        return new FixSslStreamSourceConduit(factory, exchange.getConnection());
                    }
                });
            }
            handler.handleRequest(exchange);
        }

    }

    private static class FixSslStreamSourceConduit extends AbstractStreamSourceConduit<StreamSourceConduit> {
        static final Logger log = Logger.getLogger(FixSslStreamSourceConduit.class);

        static final boolean CHECK_COUNT = false;
        static final int MAX_ZERO_READ_COUNT = 20;

        static final boolean CHECK_TIME = true;
        static final long MAX_ZERO_TIME = 5 * 1000l; // 5 seconds

        final ServerConnection serverConnection;

        public FixSslStreamSourceConduit(ConduitFactory<StreamSourceConduit> factory, ServerConnection serverConnection) {
            super(factory.create());
            this.serverConnection = serverConnection;

            log.debug("FixSslStreamSourceConduit : has been used for " + serverConnection);
        }

        int zeroCount = 0;
        long lastNonZeroTime = 0l;

        @Override
        public long transferTo(long position, long count, FileChannel target) throws IOException {
            long ret = super.transferTo(position, count, target);
            handleReturnValue(ret);
            return ret;
        }


        @Override
        public long transferTo(long count, ByteBuffer throughBuffer, StreamSinkChannel target) throws IOException {
            long ret = super.transferTo(count, throughBuffer, target);
            handleReturnValue(ret);
            return ret;
        }

        @Override
        public int read(ByteBuffer dst) throws IOException {
            int ret = super.read(dst);
            handleReturnValue(ret);
            return ret;
        }

        @Override
        public long read(ByteBuffer[] dsts, int offs, int len) throws IOException {
            long ret = super.read(dsts, offs, len);
            handleReturnValue(ret);
            return ret;
        }

        private void handleReturnValue(long ret) throws IOException {
            long nowTime = System.currentTimeMillis();

            if (lastNonZeroTime == 0l || ret > 0) {
                zeroCount = 0;
                lastNonZeroTime = nowTime;
            }
            else if(ret == 0) {
                zeroCount++;

                if(( CHECK_TIME && (lastNonZeroTime + MAX_ZERO_TIME < nowTime)) || ( CHECK_COUNT && (zeroCount > MAX_ZERO_READ_COUNT))) {
                    log.info("Connection will be closed due to excessive zero reads : " + zeroCount + " : " + serverConnection);
                    try {
                        terminateReads();
                    } finally {
                        serverConnection.close();
                    }
                }
            }
        }
    }
}