Embedded netty

.gitignore

@@ -1,3 +1,4 @@
+databases/*
 out/*
 generate_bundles.rb
 .cache

.travis.yml

@@ -1,13 +1,19 @@
 language: scala
 scala:
   - 2.10.4
-  - 2.11.0
+  - 2.11.7
 jdk:
   - oraclejdk7
-  - openjdk7
+  - oraclejdk8
 services:
   - postgresql
   - mysql
+cache:
+  directories:
+    - vendor/bundle
+    - $HOME/.m2
+    - $HOME/.ivy2
+    - $HOME/.sbt
 before_script:
   - ./script/prepare_build.sh
 

CHANGELOG.md

@@ -25,7 +25,14 @@
 
 # Changelog
 
-## 0.2.17 - in progresss
+## 0.2.18 - 2015-08-08
+
+* Timeouts implemented queries for MySQL and PostgreSQL - @lifey - #147
+
+## 0.2.17 - 2015-07-13
+
+* Fixed pool leak issue - @haski
+* Fixed date time formatting issue - #142
 
 ## 0.2.16 - 2015-01-04
 

Procfile

@@ -1,2 +1,2 @@
-postgresql: postgres -D /Users/mauricio/databases/postgresql
+postgresql: postgres -D databases/postgresql
 mysql: mysqld --log-warnings --console

copynetty.sh

@@ -16,4 +16,7 @@ rm -rf db-async-common/src/main/java/com/github/mauricio/netty/bootstrap
 cp -r netty/transport/src/main/java/io/netty/bootstrap db-async-common/src/main/java/io/netty
 
 rm -rf db-async-common/src/main/java/com/github/mauricio/netty/channel
-cp -r netty/transport/src/main/java/io/netty/channel db-async-common/src/main/java/io/netty
+cp -r netty/transport/src/main/java/io/netty/channel db-async-common/src/main/java/io/netty
+
+cp -r db-async-common/src/main/java/io/netty/* db-async-common/src/main/java/com/github/mauricio/netty
+rm -rf db-async-common/src/main/java/io/netty

db-async-common/src/main/java/com/github/mauricio/netty/bootstrap/AbstractBootstrap.java

@@ -46,7 +46,7 @@
  */
 public abstract class AbstractBootstrap<B extends AbstractBootstrap<B, C>, C extends Channel> implements Cloneable {
 
-    private volatile EventLoopGroup group;
+    volatile EventLoopGroup group;
     private volatile ChannelFactory<? extends C> channelFactory;
     private volatile SocketAddress localAddress;
     private final Map<ChannelOption<?>, Object> options = new LinkedHashMap<ChannelOption<?>, Object>();
@@ -71,7 +71,7 @@ public abstract class AbstractBootstrap<B extends AbstractBootstrap<B, C>, C ext
     }
 
     /**
-     * The {@link EventLoopGroup} which is used to handle all the events for the to-be-creates
+     * The {@link EventLoopGroup} which is used to handle all the events for the to-be-created
      * {@link Channel}
      */
     @SuppressWarnings("unchecked")
@@ -174,6 +174,7 @@ public <T> B option(ChannelOption<T> option, T value) {
      * Allow to specify an initial attribute of the newly created {@link Channel}.  If the {@code value} is
      * {@code null}, the attribute of the specified {@code key} is removed.
      */
+    @SuppressWarnings("unchecked")
     public <T> B attr(AttributeKey<T> key, T value) {
         if (key == null) {
             throw new NullPointerException("key");
@@ -187,10 +188,7 @@ public <T> B attr(AttributeKey<T> key, T value) {
                 attrs.put(key, value);
             }
         }
-
-        @SuppressWarnings("unchecked")
-        B b = (B) this;
-        return b;
+        return (B) this;
     }
 
     /**
@@ -277,7 +275,7 @@ private ChannelFuture doBind(final SocketAddress localAddress) {
         }
 
         if (regFuture.isDone()) {
-            // At this point we know that the registration was complete and succesful.
+            // At this point we know that the registration was complete and successful.
             ChannelPromise promise = channel.newPromise();
             doBind0(regFuture, channel, localAddress, promise);
             return promise;

db-async-common/src/main/java/com/github/mauricio/netty/bootstrap/Bootstrap.java

@@ -21,6 +21,7 @@
 import com.github.mauricio.netty.channel.ChannelOption;
 import com.github.mauricio.netty.channel.ChannelPipeline;
 import com.github.mauricio.netty.channel.ChannelPromise;
+import com.github.mauricio.netty.channel.EventLoopGroup;
 import com.github.mauricio.netty.util.AttributeKey;
 import com.github.mauricio.netty.util.internal.logging.InternalLogger;
 import com.github.mauricio.netty.util.internal.logging.InternalLoggerFactory;
@@ -216,6 +217,17 @@ public Bootstrap clone() {
         return new Bootstrap(this);
     }
 
+    /**
+     * Returns a deep clone of this bootstrap which has the identical configuration except that it uses
+     * the given {@link EventLoopGroup}. This method is useful when making multiple {@link Channel}s with similar
+     * settings.
+     */
+    public Bootstrap clone(EventLoopGroup group) {
+        Bootstrap bs = new Bootstrap(this);
+        bs.group = group;
+        return bs;
+    }
+
     @Override
     public String toString() {
         if (remoteAddress == null) {

db-async-common/src/main/java/com/github/mauricio/netty/bootstrap/ServerBootstrap.java

@@ -75,7 +75,7 @@ public ServerBootstrap group(EventLoopGroup group) {
 
     /**
      * Set the {@link EventLoopGroup} for the parent (acceptor) and the child (client). These
-     * {@link EventLoopGroup}'s are used to handle all the events and IO for {@link SocketChannel} and
+     * {@link EventLoopGroup}'s are used to handle all the events and IO for {@link ServerChannel} and
      * {@link Channel}'s.
      */
     public ServerBootstrap group(EventLoopGroup parentGroup, EventLoopGroup childGroup) {

db-async-common/src/main/java/com/github/mauricio/netty/buffer/AbstractByteBuf.java

@@ -1178,4 +1178,8 @@ protected final void ensureAccessible() {
             throw new IllegalReferenceCountException(0);
         }
     }
+
+    void discardMarks() {
+        markedReaderIndex = markedWriterIndex = 0;
+    }
 }

db-async-common/src/main/java/com/github/mauricio/netty/buffer/AbstractDerivedByteBuf.java

@@ -47,12 +47,20 @@ public final ByteBuf retain(int increment) {
 
     @Override
     public final boolean release() {
-        return unwrap().release();
+        if (unwrap().release()) {
+            deallocate();
+            return true;
+        }
+        return false;
     }
 
     @Override
     public final boolean release(int decrement) {
-        return unwrap().release(decrement);
+        if (unwrap().release(decrement)) {
+            deallocate();
+            return true;
+        }
+        return false;
     }
 
     @Override
@@ -64,4 +72,9 @@ public ByteBuffer internalNioBuffer(int index, int length) {
     public ByteBuffer nioBuffer(int index, int length) {
         return unwrap().nioBuffer(index, length);
     }
+
+    /**
+     * Called when the wrapped {@link ByteBuf} was released due calling of {@link #release()} or {@link #release(int)}.
+     */
+    protected void deallocate() { }
 }

db-async-common/src/main/java/com/github/mauricio/netty/buffer/ByteBuf.java

@@ -200,6 +200,9 @@
  * <p>
  * In case a completely fresh copy of an existing buffer is required, please
  * call {@link #copy()} method instead.
+ * <p>
+ * Also be aware that obtaining derived buffers will NOT call {@link #retain()} and so the
+ * reference count will NOT be increased.
  *
  * <h3>Conversion to existing JDK types</h3>
  *
@@ -1190,6 +1193,9 @@ public abstract class ByteBuf implements ReferenceCounted, Comparable<ByteBuf> {
      * Returns a new slice of this buffer's sub-region starting at the current
      * {@code readerIndex} and increases the {@code readerIndex} by the size
      * of the new slice (= {@code length}).
+     * <p>
+     * Also be aware that this method will NOT call {@link #retain()} and so the
+     * reference count will NOT be increased.
      *
      * @param length the size of the new slice
      *
@@ -1660,6 +1666,9 @@ public abstract class ByteBuf implements ReferenceCounted, Comparable<ByteBuf> {
      * identical to {@code buf.slice(buf.readerIndex(), buf.readableBytes())}.
      * This method does not modify {@code readerIndex} or {@code writerIndex} of
      * this buffer.
+     * <p>
+     * Also be aware that this method will NOT call {@link #retain()} and so the
+     * reference count will NOT be increased.
      */
     public abstract ByteBuf slice();
 
@@ -1669,6 +1678,9 @@ public abstract class ByteBuf implements ReferenceCounted, Comparable<ByteBuf> {
      * they maintain separate indexes and marks.
      * This method does not modify {@code readerIndex} or {@code writerIndex} of
      * this buffer.
+     * <p>
+     * Also be aware that this method will NOT call {@link #retain()} and so the
+     * reference count will NOT be increased.
      */
     public abstract ByteBuf slice(int index, int length);
 
@@ -1679,6 +1691,9 @@ public abstract class ByteBuf implements ReferenceCounted, Comparable<ByteBuf> {
      * This method is identical to {@code buf.slice(0, buf.capacity())}.
      * This method does not modify {@code readerIndex} or {@code writerIndex} of
      * this buffer.
+     * <p>
+     * The reader and writer marks will not be duplicated. Also be aware that this method will
+     * NOT call {@link #retain()} and so the reference count will NOT be increased.
      */
     public abstract ByteBuf duplicate();