From dab2a76eae357a0e86522d820fec5c32a40c038b Mon Sep 17 00:00:00 2001 From: Simon Willnauer Date: Tue, 30 Jun 2015 16:33:23 +0200 Subject: [PATCH] Cut over to writeable for TransportAddress This will allow us to move away from reflection hacks to serialize transport exceptions. --- .../transport/DummyTransportAddress.java | 5 ++- .../transport/InetSocketTransportAddress.java | 44 +++++++++++-------- .../transport/LocalTransportAddress.java | 11 +++-- .../common/transport/TransportAddress.java | 3 +- .../TransportAddressSerializers.java | 25 ++++------- 5 files changed, 46 insertions(+), 42 deletions(-) diff --git a/core/src/main/java/org/elasticsearch/common/transport/DummyTransportAddress.java b/core/src/main/java/org/elasticsearch/common/transport/DummyTransportAddress.java index f86bb01de76..47f089a1e14 100644 --- a/core/src/main/java/org/elasticsearch/common/transport/DummyTransportAddress.java +++ b/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 diff --git a/core/src/main/java/org/elasticsearch/common/transport/InetSocketTransportAddress.java b/core/src/main/java/org/elasticsearch/common/transport/InetSocketTransportAddress.java index 027f42a58cd..a13e24f3c3b 100644 --- a/core/src/main/java/org/elasticsearch/common/transport/InetSocketTransportAddress.java +++ b/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 { - 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()); - } + public TransportAddress readFrom(StreamInput in) throws IOException { + return new InetSocketTransportAddress(in); } @Override diff --git a/core/src/main/java/org/elasticsearch/common/transport/LocalTransportAddress.java b/core/src/main/java/org/elasticsearch/common/transport/LocalTransportAddress.java index 6a8bc082b6e..8935275e222 100644 --- a/core/src/main/java/org/elasticsearch/common/transport/LocalTransportAddress.java +++ b/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 { - id = in.readString(); + public LocalTransportAddress readFrom(StreamInput in) throws IOException { + return new LocalTransportAddress(in); } @Override diff --git a/core/src/main/java/org/elasticsearch/common/transport/TransportAddress.java b/core/src/main/java/org/elasticsearch/common/transport/TransportAddress.java index 1338b3bc1d8..c5051fadbe6 100644 --- a/core/src/main/java/org/elasticsearch/common/transport/TransportAddress.java +++ b/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 { short uniqueAddressTypeId(); diff --git a/core/src/main/java/org/elasticsearch/common/transport/TransportAddressSerializers.java b/core/src/main/java/org/elasticsearch/common/transport/TransportAddressSerializers.java index f3aa0941d46..9bfbcaaf509 100644 --- a/core/src/main/java/org/elasticsearch/common/transport/TransportAddressSerializers.java +++ b/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> addressConstructors = ImmutableMap.of(); + private static ImmutableMap ADDRESS_REGISTRY = ImmutableMap.of(); static { try { addAddressType(DummyTransportAddress.INSTANCE); - addAddressType(new InetSocketTransportAddress()); - addAddressType(new LocalTransportAddress()); + addAddressType(InetSocketTransportAddress.PROTO); + 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 constructor = address.getClass().getDeclaredConstructor(); - constructor.setAccessible(true); - addressConstructors = newMapBuilder(addressConstructors).put(address.uniqueAddressTypeId(), constructor).immutableMap(); + ADDRESS_REGISTRY = newMapBuilder(ADDRESS_REGISTRY).put(address.uniqueAddressTypeId(), address).immutableMap(); } public static TransportAddress addressFromStream(StreamInput input) throws IOException { short addressUniqueId = input.readShort(); - Constructor constructor = addressConstructors.get(addressUniqueId); - if (constructor == null) { + TransportAddress addressType = ADDRESS_REGISTRY.get(addressUniqueId); + if (addressType == null) { throw new IOException("No transport address mapped to [" + addressUniqueId + "]"); } - TransportAddress address; - try { - address = constructor.newInstance(); - } catch (Exception e) { - throw new IOException("Failed to create class with constructor [" + constructor + "]", e); - } - address.readFrom(input); - return address; + return addressType.readFrom(input); } public static void addressToStream(StreamOutput out, TransportAddress address) throws IOException {