Merge pull request #6945 from alrossi/fix/7.2/xroot-outbound-errors

dcache-xroot: handle outbound errors on channel promise

modules/dcache-xrootd/src/main/java/org/dcache/xrootd/OutboundExceptionHandler.java

@@ -0,0 +1,36 @@
+/* dCache - http://www.dcache.org/
+ *
+ * Copyright (C) 2014 Deutsches Elektronen-Synchrotron
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package org.dcache.xrootd;
+
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelOutboundHandlerAdapter;
+import io.netty.channel.ChannelPromise;
+
+/**
+ *  Should be added at the beginning of each pipeline to ensure that outbound errors
+ *  on the channel promise are handled.
+ */
+public class OutboundExceptionHandler extends ChannelOutboundHandlerAdapter {
+    @Override
+    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
+          throws Exception {
+        promise.addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE);
+        super.write(ctx, msg, promise);
+    }
+}

modules/dcache-xrootd/src/main/java/org/dcache/xrootd/door/NettyXrootdServer.java

@@ -35,6 +35,7 @@
 import java.util.concurrent.TimeUnit;
 import org.dcache.util.CDCThreadFactory;
 import org.dcache.util.NDC;
+import org.dcache.xrootd.OutboundExceptionHandler;
 import org.dcache.xrootd.core.XrootdAuthenticationHandler;
 import org.dcache.xrootd.core.XrootdDecoder;
 import org.dcache.xrootd.core.XrootdEncoder;
@@ -211,6 +212,7 @@ protected void initChannel(Channel ch) throws Exception {
                             Longs.toByteArray(sessionCounter.next()));
 
                       ChannelPipeline pipeline = ch.pipeline();
+                      pipeline.addLast("outerrors", new OutboundExceptionHandler());
                       pipeline.addLast("session", new SessionHandler(session));
                       if (_expectProxyProtocol) {
                           pipeline.addLast("haproxy", new HAProxyMessageDecoder());

modules/dcache-xrootd/src/main/java/org/dcache/xrootd/pool/XrootdTransferService.java

@@ -67,6 +67,7 @@
 import org.dcache.util.NetworkUtils;
 import org.dcache.vehicles.XrootdDoorAdressInfoMessage;
 import org.dcache.vehicles.XrootdProtocolInfo;
+import org.dcache.xrootd.OutboundExceptionHandler;
 import org.dcache.xrootd.core.XrootdDecoder;
 import org.dcache.xrootd.core.XrootdEncoder;
 import org.dcache.xrootd.core.XrootdHandshakeHandler;
@@ -397,7 +398,7 @@ protected void initChannel(Channel ch) throws Exception {
         super.initChannel(ch);
 
         ChannelPipeline pipeline = ch.pipeline();
-
+        pipeline.addLast("outerrors", new OutboundExceptionHandler());
         pipeline.addLast("handshake",
               new XrootdHandshakeHandler(XrootdProtocol.DATA_SERVER));
         pipeline.addLast("encoder", new XrootdEncoder());