Cut over to writeable for TransportAddress

This will allow us to move away from reflection hacks to serialize
transport exceptions.

core/src/main/java/org/elasticsearch/common/transport/DummyTransportAddress.java

@@ -31,7 +31,7 @@ public class DummyTransportAddress implements TransportAddress {
 
     public static final DummyTransportAddress INSTANCE = new DummyTransportAddress();
 
-    DummyTransportAddress() {
+    private DummyTransportAddress() {
     }
 
     @Override
@@ -45,7 +45,8 @@ public class DummyTransportAddress implements TransportAddress {
     }
 
     @Override
-    public void readFrom(StreamInput in) throws IOException {
+    public DummyTransportAddress readFrom(StreamInput in) throws IOException {
+        return INSTANCE;
     }
 
     @Override

core/src/main/java/org/elasticsearch/common/transport/InetSocketTransportAddress.java

@@ -42,10 +42,31 @@ public class InetSocketTransportAddress implements TransportAddress {
         return resolveAddress;
     }
 
-    private InetSocketAddress address;
+    public static final InetSocketTransportAddress PROTO = new InetSocketTransportAddress();
 
-    InetSocketTransportAddress() {
+    private final InetSocketAddress address;
 
+    public InetSocketTransportAddress(StreamInput in) throws IOException {
+        if (in.readByte() == 0) {
+            int len = in.readByte();
+            byte[] a = new byte[len]; // 4 bytes (IPv4) or 16 bytes (IPv6)
+            in.readFully(a);
+            InetAddress inetAddress;
+            if (len == 16) {
+                int scope_id = in.readInt();
+                inetAddress = Inet6Address.getByAddress(null, a, scope_id);
+            } else {
+                inetAddress = InetAddress.getByAddress(a);
+            }
+            int port = in.readInt();
+            this.address = new InetSocketAddress(inetAddress, port);
+        } else {
+            this.address = new InetSocketAddress(in.readString(), in.readInt());
+        }
+    }
+
+    private InetSocketTransportAddress() {
+        address = null;
     }
 
     public InetSocketTransportAddress(String hostname, int port) {
@@ -76,23 +97,8 @@ public class InetSocketTransportAddress implements TransportAddress {
     }
 
     @Override
-    public void readFrom(StreamInput in) throws IOException {
+    public TransportAddress readFrom(StreamInput in) throws IOException {
-        if (in.readByte() == 0) {
+        return new InetSocketTransportAddress(in);
-            int len = in.readByte();
-            byte[] a = new byte[len]; // 4 bytes (IPv4) or 16 bytes (IPv6)
-            in.readFully(a);
-            InetAddress inetAddress;
-            if (len == 16) {
-                int scope_id = in.readInt();
-                inetAddress = Inet6Address.getByAddress(null, a, scope_id);
-            } else {
-                inetAddress = InetAddress.getByAddress(a);
-            }
-            int port = in.readInt();
-            this.address = new InetSocketAddress(inetAddress, port);
-        } else {
-            this.address = new InetSocketAddress(in.readString(), in.readInt());
-        }
     }
 
     @Override

core/src/main/java/org/elasticsearch/common/transport/LocalTransportAddress.java

@@ -21,6 +21,8 @@ package org.elasticsearch.common.transport;
 
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.transport.local.LocalTransport;
 
 import java.io.IOException;
 
@@ -29,9 +31,12 @@ import java.io.IOException;
  */
 public class LocalTransportAddress implements TransportAddress {
 
+    public static final LocalTransportAddress PROTO = new LocalTransportAddress("_na");
+
     private String id;
 
-    LocalTransportAddress() {
+    public LocalTransportAddress(StreamInput in) throws IOException {
+        id = in.readString();
     }
 
     public LocalTransportAddress(String id) {
@@ -53,8 +58,8 @@ public class LocalTransportAddress implements TransportAddress {
     }
 
     @Override
-    public void readFrom(StreamInput in) throws IOException {
+    public LocalTransportAddress readFrom(StreamInput in) throws IOException {
-        id = in.readString();
+        return new LocalTransportAddress(in);
     }
 
     @Override

core/src/main/java/org/elasticsearch/common/transport/TransportAddress.java

@@ -20,12 +20,13 @@
 package org.elasticsearch.common.transport;
 
 import org.elasticsearch.common.io.stream.Streamable;
+import org.elasticsearch.common.io.stream.Writeable;
 
 
 /**
  *
  */
-public interface TransportAddress extends Streamable {
+public interface TransportAddress extends Writeable<TransportAddress> {
 
     short uniqueAddressTypeId();
 

core/src/main/java/org/elasticsearch/common/transport/TransportAddressSerializers.java

@@ -42,41 +42,32 @@ public abstract class TransportAddressSerializers {
 
     private static final ESLogger logger = Loggers.getLogger(TransportAddressSerializers.class);
 
-    private static ImmutableMap<Short, Constructor<? extends TransportAddress>> addressConstructors = ImmutableMap.of();
+    private static ImmutableMap<Short, TransportAddress> ADDRESS_REGISTRY = ImmutableMap.of();
 
     static {
         try {
             addAddressType(DummyTransportAddress.INSTANCE);
-            addAddressType(new InetSocketTransportAddress());
+            addAddressType(InetSocketTransportAddress.PROTO);
-            addAddressType(new LocalTransportAddress());
+            addAddressType(LocalTransportAddress.PROTO);
         } catch (Exception e) {
             logger.warn("Failed to add InetSocketTransportAddress", e);
         }
     }
 
     public static synchronized void addAddressType(TransportAddress address) throws Exception {
-        if (addressConstructors.containsKey(address.uniqueAddressTypeId())) {
+        if (ADDRESS_REGISTRY.containsKey(address.uniqueAddressTypeId())) {
             throw new IllegalStateException("Address [" + address.uniqueAddressTypeId() + "] already bound");
         }
-        Constructor<? extends TransportAddress> constructor = address.getClass().getDeclaredConstructor();
+        ADDRESS_REGISTRY = newMapBuilder(ADDRESS_REGISTRY).put(address.uniqueAddressTypeId(), address).immutableMap();
-        constructor.setAccessible(true);
-        addressConstructors = newMapBuilder(addressConstructors).put(address.uniqueAddressTypeId(), constructor).immutableMap();
     }
 
     public static TransportAddress addressFromStream(StreamInput input) throws IOException {
         short addressUniqueId = input.readShort();
-        Constructor<? extends TransportAddress> constructor = addressConstructors.get(addressUniqueId);
+        TransportAddress addressType = ADDRESS_REGISTRY.get(addressUniqueId);
-        if (constructor == null) {
+        if (addressType == null) {
             throw new IOException("No transport address mapped to [" + addressUniqueId + "]");
         }
-        TransportAddress address;
+        return addressType.readFrom(input);
-        try {
-            address = constructor.newInstance();
-        } catch (Exception e) {
-            throw new IOException("Failed to create class with constructor [" + constructor + "]", e);
-        }
-        address.readFrom(input);
-        return address;
     }
 
     public static void addressToStream(StreamOutput out, TransportAddress address) throws IOException {