Repository: relayrides/pushy (java)

Issue:
Rare race condition when getting "channel ready" promise
As [spotted by Travis](https://travis-ci.org/relayrides/pushy/jobs/283355904), we can, in some cases, try to get the "channel is ready" promise attribute of a channel before it has actually been attached to the channel. This can happen if the server responds faster than we can attach the attribute, which is _extremely_ unlikely in real-world conditions.

Still, it'd be good to get this sorted out.

A stack trace, just in case Travis discards the logs somehow:

```
[nioEventLoopGroup-14-2] WARN io.netty.channel.DefaultChannelPipeline - An exceptionCaught() event was fired, and it reached at the tail of the pipeline. It usually means the last handler in the pipeline did not handle the exception.
java.lang.NullPointerException
	at com.turo.pushy.apns.ApnsChannelFactory$1$2.configurePipeline(ApnsChannelFactory.java:138)
	at io.netty.handler.ssl.ApplicationProtocolNegotiationHandler.userEventTriggered(ApplicationProtocolNegotiationHandler.java:92)
	at io.netty.channel.AbstractChannelHandlerContext.invokeUserEventTriggered(AbstractChannelHandlerContext.java:329)
	at io.netty.channel.AbstractChannelHandlerContext.invokeUserEventTriggered(AbstractChannelHandlerContext.java:315)
	at io.netty.channel.AbstractChannelHandlerContext.fireUserEventTriggered(AbstractChannelHandlerContext.java:307)
	at io.netty.handler.ssl.SslHandler.setHandshakeSuccess(SslHandler.java:1437)
	at io.netty.handler.ssl.SslHandler.unwrap(SslHandler.java:1275)
	at io.netty.handler.ssl.SslHandler.decodeNonJdkCompatible(SslHandler.java:1139)
	at io.netty.handler.ssl.SslHandler.decode(SslHandler.java:1164)
	at io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:489)
	at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:428)
	at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:265)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)
	at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1359)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
	at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:935)
	at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:134)
	at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:645)
	at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:580)
	at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:497)
	at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:459)
	at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:858)
	at io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:138)
	at java.lang.Thread.run(Thread.java:748)
```

Pull request:
Attach "connection ready" promise to channels at construction time
This fixes #533, a race condition that could happen because we weren't attaching an attribute to a newly-created channel until after the connection process had already begun. This change introduces a `ReflectiveChannelFactory` subclass that attaches an attribute to channels at creation time, which should solve the race condition.

Patch:
diff --git a/pushy/src/main/java/com/turo/pushy/apns/ApnsChannelFactory.java b/pushy/src/main/java/com/turo/pushy/apns/ApnsChannelFactory.java
index 1f42ea00..85f46c4e 100644
--- a/pushy/src/main/java/com/turo/pushy/apns/ApnsChannelFactory.java
+++ b/pushy/src/main/java/com/turo/pushy/apns/ApnsChannelFactory.java
@@ -77,7 +77,6 @@ class ApnsChannelFactory implements PooledObjectFactory<Channel>, Closeable {
 
         this.bootstrapTemplate = new Bootstrap();
         this.bootstrapTemplate.group(eventLoopGroup);
-        this.bootstrapTemplate.channel(SocketChannelClassUtil.getSocketChannelClass(this.bootstrapTemplate.config().group()));
         this.bootstrapTemplate.option(ChannelOption.TCP_NODELAY, true);
         this.bootstrapTemplate.remoteAddress(apnsServerAddress);
 
@@ -169,12 +168,18 @@ class ApnsChannelFactory implements PooledObjectFactory<Channel>, Closeable {
             }
         });
 
+
         this.bootstrapTemplate.config().group().schedule(new Runnable() {
 
             @Override
             public void run() {
-                final ChannelFuture connectFuture = ApnsChannelFactory.this.bootstrapTemplate.clone().connect();
-                connectFuture.channel().attr(CHANNEL_READY_PROMISE_ATTRIBUTE_KEY).set(channelReadyPromise);
+
+                final Bootstrap bootstrap = ApnsChannelFactory.this.bootstrapTemplate.clone()
+                        .channelFactory(new AugmentingReflectiveChannelFactory<>(
+                                SocketChannelClassUtil.getSocketChannelClass(ApnsChannelFactory.this.bootstrapTemplate.config().group()),
+                                CHANNEL_READY_PROMISE_ATTRIBUTE_KEY, channelReadyPromise));
+
+                final ChannelFuture connectFuture = bootstrap.connect();
 
                 connectFuture.addListener(new GenericFutureListener<ChannelFuture>() {
 
diff --git a/pushy/src/main/java/com/turo/pushy/apns/AugmentingReflectiveChannelFactory.java b/pushy/src/main/java/com/turo/pushy/apns/AugmentingReflectiveChannelFactory.java
new file mode 100644
index 00000000..46bdb1f5
--- /dev/null
+++ b/pushy/src/main/java/com/turo/pushy/apns/AugmentingReflectiveChannelFactory.java
@@ -0,0 +1,61 @@
+/*
+ * Copyright (c) 2013-2017 Turo
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+ * THE SOFTWARE.
+ */
+
+package com.turo.pushy.apns;
+
+import io.netty.channel.Channel;
+import io.netty.channel.ReflectiveChannelFactory;
+import io.netty.util.AttributeKey;
+
+import java.util.Objects;
+
+/**
+ * A channel factory that attaches a single attribute to all newly-created channels.
+ *
+ * @param <T> the type of channel created by this factory
+ * @param <A> the type of the attribute attached to newly-constructed channels
+ *
+ * @author <a href="https://github.com/jchambers">Jon Chambers</a>
+ */
+class AugmentingReflectiveChannelFactory<T extends Channel, A> extends ReflectiveChannelFactory<T> {
+
+    private final AttributeKey<A> attributeKey;
+    private final A attributeValue;
+
+    AugmentingReflectiveChannelFactory(final Class<? extends T> channelClass, final AttributeKey<A> attributeKey, final A attributeValue) {
+        super(channelClass);
+
+        Objects.requireNonNull(attributeKey, "Attribute key must not be null.");
+
+        this.attributeKey = attributeKey;
+        this.attributeValue = attributeValue;
+    }
+
+    @Override
+    public T newChannel() {
+        final T channel = super.newChannel();
+
+        channel.attr(this.attributeKey).set(this.attributeValue);
+
+        return channel;
+    }
+}


Test patch:
diff --git a/pushy/src/test/java/com/turo/pushy/apns/AugmentingReflectiveChannelFactoryTest.java b/pushy/src/test/java/com/turo/pushy/apns/AugmentingReflectiveChannelFactoryTest.java
new file mode 100644
index 00000000..5f8937ce
--- /dev/null
+++ b/pushy/src/test/java/com/turo/pushy/apns/AugmentingReflectiveChannelFactoryTest.java
@@ -0,0 +1,50 @@
+/*
+ * Copyright (c) 2013-2017 Turo
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+ * THE SOFTWARE.
+ */
+
+package com.turo.pushy.apns;
+
+import io.netty.channel.local.LocalChannel;
+import io.netty.util.AttributeKey;
+import org.junit.Test;
+
+import static org.junit.Assert.assertTrue;
+
+public class AugmentingReflectiveChannelFactoryTest {
+
+    @Test(expected = NullPointerException.class)
+    public void testAugmentingReflectiveChannelFactoryNullAttributeKey() {
+        new AugmentingReflectiveChannelFactory<>(LocalChannel.class, null, null);
+    }
+
+    @Test
+    public void testNewChannel() {
+        final AttributeKey<String> attributeKey = AttributeKey.valueOf(getClass(), "attributeKey");
+
+        final AugmentingReflectiveChannelFactory<LocalChannel, String> factory =
+                new AugmentingReflectiveChannelFactory<>(LocalChannel.class, attributeKey, "Test!");
+
+        final LocalChannel channel = factory.newChannel();
+
+        assertTrue("Newly-created channels should have attribute provided to factory.",
+                channel.hasAttr(attributeKey));
+    }
+}