[#957] Netty/Transport/RXTX: Add an optional wait time after opening the serial port but before configuration to allow the serial microcontroller to reset itself after opening.

transport-rxtx/src/main/java/io/netty/channel/rxtx/DefaultRxtxChannelConfig.java

@@ -34,6 +34,7 @@ final class DefaultRxtxChannelConfig extends DefaultChannelConfig implements Rxt
     private volatile Stopbits stopbits = Stopbits.STOPBITS_1;
     private volatile Databits databits = Databits.DATABITS_8;
     private volatile Paritybit paritybit = Paritybit.NONE;
+    private volatile int waitTime;
 
     public DefaultRxtxChannelConfig(RxtxChannel channel) {
         super(channel);
@@ -41,7 +42,7 @@ final class DefaultRxtxChannelConfig extends DefaultChannelConfig implements Rxt
 
     @Override
     public Map<ChannelOption<?>, Object> getOptions() {
-        return getOptions(super.getOptions(), BAUD_RATE, DTR, RTS, STOP_BITS, DATA_BITS, PARITY_BIT);
+        return getOptions(super.getOptions(), BAUD_RATE, DTR, RTS, STOP_BITS, DATA_BITS, PARITY_BIT, WAIT_TIME);
     }
 
     @SuppressWarnings("unchecked")
@@ -65,6 +66,9 @@ final class DefaultRxtxChannelConfig extends DefaultChannelConfig implements Rxt
         if (option == PARITY_BIT) {
             return (T) getParitybit();
         }
+        if (option == WAIT_TIME) {
+            return (T) Integer.valueOf(getWaitTimeMillis());
+        }
         return super.getOption(option);
     }
 
@@ -84,6 +88,8 @@ final class DefaultRxtxChannelConfig extends DefaultChannelConfig implements Rxt
             setDatabits((Databits) value);
         } else if (option == PARITY_BIT) {
             setParitybit((Paritybit) value);
+        } else if (option == WAIT_TIME) {
+            setWaitTimeMillis((Integer) value);
         } else {
             return super.setOption(option, value);
         }
@@ -156,6 +162,20 @@ final class DefaultRxtxChannelConfig extends DefaultChannelConfig implements Rxt
         return this;
     }
 
+    @Override
+    public int getWaitTimeMillis() {
+        return waitTime;
+    }
+
+    @Override
+    public RxtxChannelConfig setWaitTimeMillis(final int waitTimeMillis) {
+        if (waitTimeMillis < 0) {
+            throw new IllegalArgumentException("Wait time must be >= 0");
+        }
+        waitTime = waitTimeMillis;
+        return this;
+    }
+
     @Override
     public RxtxChannelConfig setConnectTimeoutMillis(int connectTimeoutMillis) {
         return (RxtxChannelConfig) super.setConnectTimeoutMillis(connectTimeoutMillis);

transport-rxtx/src/main/java/io/netty/channel/rxtx/RxtxChannel.java

@@ -15,14 +15,17 @@
  */
 package io.netty.channel.rxtx;
 
-import gnu.io.CommPort;
+import static io.netty.channel.rxtx.RxtxChannelOption.*;
-import gnu.io.CommPortIdentifier;
+
-import gnu.io.SerialPort;
+import io.netty.channel.ChannelPromise;
 import io.netty.channel.socket.oio.OioByteStreamChannel;
 
 import java.net.SocketAddress;
+import java.util.concurrent.TimeUnit;
 
-import static io.netty.channel.rxtx.RxtxChannelOption.*;
+import gnu.io.CommPort;
+import gnu.io.CommPortIdentifier;
+import gnu.io.SerialPort;
 
 /**
  * A channel to a serial device using the RXTX library.
@@ -53,6 +56,11 @@ public class RxtxChannel extends OioByteStreamChannel {
         return open;
     }
 
+    @Override
+    protected AbstractUnsafe newUnsafe() {
+        return new RxtxUnsafe();
+    }
+
     @Override
     protected void doConnect(SocketAddress remoteAddress, SocketAddress localAddress) throws Exception {
         RxtxDeviceAddress remote = (RxtxDeviceAddress) remoteAddress;
@@ -62,6 +70,9 @@ public class RxtxChannel extends OioByteStreamChannel {
         deviceAddress = remote;
 
         serialPort = (SerialPort) commPort;
+    }
+
+    protected void doInit() throws Exception {
         serialPort.setSerialPortParams(
             config().getOption(BAUD_RATE),
             config().getOption(DATA_BITS).value(),
@@ -117,4 +128,57 @@ public class RxtxChannel extends OioByteStreamChannel {
             }
         }
     }
+
+    private final class RxtxUnsafe extends AbstractUnsafe {
+        @Override
+        public void connect(
+                final SocketAddress remoteAddress,
+                final SocketAddress localAddress, final ChannelPromise promise) {
+            if (eventLoop().inEventLoop()) {
+                if (!ensureOpen(promise)) {
+                    return;
+                }
+
+                try {
+                    final boolean wasActive = isActive();
+                    doConnect(remoteAddress, localAddress);
+
+                    int waitTime = config().getOption(WAIT_TIME);
+                    if (waitTime > 0) {
+                        eventLoop().schedule(new Runnable() {
+                            @Override
+                            public void run() {
+                                try {
+                                    doInit();
+                                    promise.setSuccess();
+                                    if (!wasActive && isActive()) {
+                                        pipeline().fireChannelActive();
+                                    }
+                                } catch (Throwable t) {
+                                    promise.setFailure(t);
+                                    closeIfClosed();
+                                }
+                            }
+                       }, waitTime, TimeUnit.MILLISECONDS);
+                    } else {
+                        doInit();
+                        promise.setSuccess();
+                        if (!wasActive && isActive()) {
+                            pipeline().fireChannelActive();
+                        }
+                    }
+                } catch (Throwable t) {
+                    promise.setFailure(t);
+                    closeIfClosed();
+                }
+            } else {
+                eventLoop().execute(new Runnable() {
+                    @Override
+                    public void run() {
+                        connect(remoteAddress, localAddress, promise);
+                    }
+                });
+            }
+        }
+    }
 }

transport-rxtx/src/main/java/io/netty/channel/rxtx/RxtxChannelConfig.java

@@ -42,6 +42,8 @@ import io.netty.channel.ChannelConfig;
  * <td>{@link io.netty.channel.rxtx.RxtxChannelOption#DATA_BITS}</td><td>{@link #setDatabits(Databits)}</td>
  * </tr><tr>
  * <td>{@link io.netty.channel.rxtx.RxtxChannelOption#PARITY_BIT}</td><td>{@link #setParitybit(Paritybit)}</td>
+ * </tr><tr>
+ * <td>{@link io.netty.channel.rxtx.RxtxChannelOption#WAIT_TIME}</td><td>{@link #setWaitTimeMillis(int)}</td>
  * </tr>
  * </table>
  */
@@ -239,6 +241,23 @@ public interface RxtxChannelConfig extends ChannelConfig {
      */
     RxtxChannelConfig setRts(boolean rts);
 
+    /**
+     * @return The number of milliseconds to wait between opening the serial port and
+     *     initialising.
+     */
+    int getWaitTimeMillis();
+
+    /**
+     * Sets the time to wait after opening the serial port and before sending it any
+     * configuration information or data. A value of 0 indicates that no waiting should
+     * occur.
+     *
+     * @param waitTimeMillis The number of milliseconds to wait, defaulting to 0 (no
+     *     wait) if unset
+     * @throws IllegalArgumentException if the supplied value is < 0
+     */
+    RxtxChannelConfig setWaitTimeMillis(int waitTimeMillis);
+
     @Override
     RxtxChannelConfig setConnectTimeoutMillis(int connectTimeoutMillis);
 

transport-rxtx/src/main/java/io/netty/channel/rxtx/RxtxChannelOption.java

@@ -42,6 +42,9 @@ public final class RxtxChannelOption<T> extends ChannelOption<T> {
     public static final RxtxChannelOption<Paritybit> PARITY_BIT =
             new RxtxChannelOption<Paritybit>("PARITY_BIT");
 
+    public static final RxtxChannelOption<Integer> WAIT_TIME =
+            new RxtxChannelOption<Integer>("WAIT_TIME");
+
     private RxtxChannelOption(String name) {
         super(name);
     }