Skip to content

Commit 257c7d1

Browse files
nikagranikagra
committed
[#596] Storing ProtocolFeatureStore in ProtocolEncoder/ProtocolDecoder. Getting read of sharing it via Channel attributes.
1 parent a824ad9 commit 257c7d1

9 files changed

Lines changed: 80 additions & 99 deletions

File tree

‎driver-core/src/main/java/com/datastax/driver/core/Connection.java‎

Lines changed: 18 additions & 40 deletions
Original file line numberDiff line numberDiff line change
@@ -84,7 +84,6 @@
8484
import java.util.Iterator;
8585
import java.util.List;
8686
import java.util.Map;
87-
import java.util.Objects;
8887
import java.util.Queue;
8988
import java.util.concurrent.ConcurrentHashMap;
9089
import java.util.concurrent.ConcurrentLinkedQueue;
@@ -156,6 +155,7 @@ enum State {
156155

157156
private final AtomicReference<Owner> ownerRef = new AtomicReference<Owner>();
158157
private final ApplicationInfo applicationInfo;
158+
private ProtocolFeatureStore protocolFeatureStore;
159159

160160
/**
161161
* Create a new connection to a Cassandra node and associate it with the given pool.
@@ -451,12 +451,10 @@ private AsyncFunction<Message.Response, Void> onOptionsResponse(
451451
return new AsyncFunction<Message.Response, Void>() {
452452
@Override
453453
public ListenableFuture<Void> apply(Message.Response response) {
454-
switch (Objects.requireNonNull(response).type) {
454+
switch (response.type) {
455455
case SUPPORTED:
456456
Supported supported = (Supported) response;
457-
ProtocolFeatureStore protocolFeatureStore =
458-
ProtocolFeatureStore.parseSupportedOptions(supported.supported);
459-
protocolFeatureStore.storeInHostAndChannel(getHost(), Connection.this.channel);
457+
protocolFeatureStore = ProtocolFeatureStore.parseSupportedOptions(supported.supported);
460458

461459
ShardingInfo.ConnectionShardingInfo shardingInfo =
462460
protocolFeatureStore.getConnectionShardingInfo();
@@ -507,7 +505,7 @@ public ListenableFuture<Void> apply(Void input) throws Exception {
507505
if (applicationInfo != null) {
508506
applicationInfo.addOption(extraOptions);
509507
}
510-
ProtocolFeatureStore protocolFeatureStore = ProtocolFeatureStore.loadFromHost(getHost());
508+
511509
if (protocolFeatureStore != null) {
512510
protocolFeatureStore.populateStartupOptions(protocolVersion, extraOptions);
513511
}
@@ -621,6 +619,7 @@ private ListenableFuture<Void> checkClusterName(
621619
new DefaultResultSetFuture(
622620
null,
623621
protocolVersion,
622+
protocolFeatureStore,
624623
new Requests.Query("select cluster_name from system.local where key = 'local'"));
625624
try {
626625
write(clusterNameFuture);
@@ -1055,6 +1054,10 @@ public int shardId() {
10551054
return shardId == null ? 0 : shardId;
10561055
}
10571056

1057+
public ProtocolFeatureStore getProtocolFeatureStore() {
1058+
return protocolFeatureStore;
1059+
}
1060+
10581061
/**
10591062
* If the connection is part of a pool, return it to the pool. The connection should generally not
10601063
* be reused after that.
@@ -1945,21 +1948,6 @@ interface DefaultResponseHandler {
19451948
}
19461949

19471950
private static class Initializer extends ChannelInitializer<SocketChannel> {
1948-
// Stateless handlers
1949-
private static final Message.ProtocolDecoder messageDecoder = new Message.ProtocolDecoder();
1950-
private static final Message.ProtocolEncoder messageEncoderV1 =
1951-
new Message.ProtocolEncoder(ProtocolVersion.V1);
1952-
private static final Message.ProtocolEncoder messageEncoderV2 =
1953-
new Message.ProtocolEncoder(ProtocolVersion.V2);
1954-
private static final Message.ProtocolEncoder messageEncoderV3 =
1955-
new Message.ProtocolEncoder(ProtocolVersion.V3);
1956-
private static final Message.ProtocolEncoder messageEncoderV4 =
1957-
new Message.ProtocolEncoder(ProtocolVersion.V4);
1958-
private static final Message.ProtocolEncoder messageEncoderV5 =
1959-
new Message.ProtocolEncoder(ProtocolVersion.V5);
1960-
private static final Message.ProtocolEncoder messageEncoderV6 =
1961-
new Message.ProtocolEncoder(ProtocolVersion.V6);
1962-
private static final Frame.Encoder frameEncoder = new Frame.Encoder();
19631951

19641952
private final ProtocolVersion protocolVersion;
19651953
private final Connection connection;
@@ -2023,7 +2011,7 @@ protected void initChannel(SocketChannel channel) throws Exception {
20232011
}
20242012

20252013
pipeline.addLast("frameDecoder", new Frame.Decoder());
2026-
pipeline.addLast("frameEncoder", frameEncoder);
2014+
pipeline.addLast("frameEncoder", new Frame.Encoder());
20272015

20282016
pipeline.addLast("framingFormatHandler", new FramingFormatHandler(connection.factory));
20292017

@@ -2036,8 +2024,9 @@ protected void initChannel(SocketChannel channel) throws Exception {
20362024
pipeline.addLast("frameCompressor", new Frame.Compressor(compressor));
20372025
}
20382026

2039-
pipeline.addLast("messageDecoder", messageDecoder);
2040-
pipeline.addLast("messageEncoder", messageEncoderFor(protocolVersion));
2027+
ProtocolFeatureStore protocolFeatureStore = connection.getProtocolFeatureStore();
2028+
pipeline.addLast("messageDecoder", new Message.ProtocolDecoder(protocolFeatureStore));
2029+
pipeline.addLast("messageEncoder", messageEncoderFor(protocolVersion, protocolFeatureStore));
20412030

20422031
pipeline.addLast("idleStateHandler", idleStateHandler);
20432032

@@ -2046,23 +2035,12 @@ protected void initChannel(SocketChannel channel) throws Exception {
20462035
nettyOptions.afterChannelInitialized(channel);
20472036
}
20482037

2049-
private Message.ProtocolEncoder messageEncoderFor(ProtocolVersion version) {
2050-
switch (version) {
2051-
case V1:
2052-
return messageEncoderV1;
2053-
case V2:
2054-
return messageEncoderV2;
2055-
case V3:
2056-
return messageEncoderV3;
2057-
case V4:
2058-
return messageEncoderV4;
2059-
case V5:
2060-
return messageEncoderV5;
2061-
case V6:
2062-
return messageEncoderV6;
2063-
default:
2064-
throw new DriverInternalError("Unsupported protocol version " + protocolVersion);
2038+
private Message.ProtocolEncoder messageEncoderFor(
2039+
ProtocolVersion version, ProtocolFeatureStore protocolFeatureStore) {
2040+
if (version.toInt() > ProtocolVersion.V6.toInt()) {
2041+
throw new DriverInternalError("Unsupported protocol version " + protocolVersion);
20652042
}
2043+
return new Message.ProtocolEncoder(version, protocolFeatureStore);
20662044
}
20672045
}
20682046

‎driver-core/src/main/java/com/datastax/driver/core/ControlConnection.java‎

Lines changed: 25 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -505,7 +505,11 @@ private Row fetchNodeInfo(Host host, Connection c)
505505
+ "'";
506506
}
507507
DefaultResultSetFuture future =
508-
new DefaultResultSetFuture(null, cluster.protocolVersion(), new Requests.Query(query));
508+
new DefaultResultSetFuture(
509+
null,
510+
cluster.protocolVersion(),
511+
c.getProtocolFeatureStore(),
512+
new Requests.Query(query));
509513
c.write(future);
510514
Row row = future.get().one();
511515
if (row != null) {
@@ -722,7 +726,10 @@ private ListenableFuture<ResultSet> selectPeersFuture(final Connection connectio
722726
if (isPeersV2) {
723727
DefaultResultSetFuture peersV2Future =
724728
new DefaultResultSetFuture(
725-
null, cluster.protocolVersion(), new Requests.Query(SELECT_PEERS_V2));
729+
null,
730+
cluster.protocolVersion(),
731+
connection.getProtocolFeatureStore(),
732+
new Requests.Query(SELECT_PEERS_V2));
726733
connection.write(peersV2Future);
727734
final SettableFuture<ResultSet> peersFuture = SettableFuture.create();
728735
// if peers v2 query fails, query peers table instead.
@@ -755,7 +762,10 @@ public void onFailure(Throwable t) {
755762
} else {
756763
DefaultResultSetFuture peersFuture =
757764
new DefaultResultSetFuture(
758-
null, cluster.protocolVersion(), new Requests.Query(SELECT_PEERS));
765+
null,
766+
cluster.protocolVersion(),
767+
connection.getProtocolFeatureStore(),
768+
new Requests.Query(SELECT_PEERS));
759769
connection.write(peersFuture);
760770
return peersFuture;
761771
}
@@ -776,7 +786,10 @@ private void refreshNodeListAndTokenMap(
776786

777787
DefaultResultSetFuture localFuture =
778788
new DefaultResultSetFuture(
779-
null, cluster.protocolVersion(), new Requests.Query(SELECT_LOCAL));
789+
null,
790+
cluster.protocolVersion(),
791+
connection.getProtocolFeatureStore(),
792+
new Requests.Query(SELECT_LOCAL));
780793
ListenableFuture<ResultSet> peersFuture = selectPeersFuture(connection);
781794
connection.write(localFuture);
782795

@@ -1075,10 +1088,16 @@ private static boolean checkSchemaAgreement(Connection connection, Cluster.Manag
10751088
throws InterruptedException, ExecutionException {
10761089
DefaultResultSetFuture peersFuture =
10771090
new DefaultResultSetFuture(
1078-
null, cluster.protocolVersion(), new Requests.Query(SELECT_SCHEMA_PEERS));
1091+
null,
1092+
cluster.protocolVersion(),
1093+
connection.getProtocolFeatureStore(),
1094+
new Requests.Query(SELECT_SCHEMA_PEERS));
10791095
DefaultResultSetFuture localFuture =
10801096
new DefaultResultSetFuture(
1081-
null, cluster.protocolVersion(), new Requests.Query(SELECT_SCHEMA_LOCAL));
1097+
null,
1098+
cluster.protocolVersion(),
1099+
connection.getProtocolFeatureStore(),
1100+
new Requests.Query(SELECT_SCHEMA_LOCAL));
10821101
connection.write(peersFuture);
10831102
connection.write(localFuture);
10841103

‎driver-core/src/main/java/com/datastax/driver/core/DefaultResultSetFuture.java‎

Lines changed: 9 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -40,12 +40,17 @@ class DefaultResultSetFuture extends AbstractFuture<ResultSet>
4040
private final SessionManager session;
4141
private final ProtocolVersion protocolVersion;
4242
private final Message.Request request;
43+
private final ProtocolFeatureStore protocolFeatureStore;
4344
private volatile RequestHandler handler;
4445

4546
DefaultResultSetFuture(
46-
SessionManager session, ProtocolVersion protocolVersion, Message.Request request) {
47+
SessionManager session,
48+
ProtocolVersion protocolVersion,
49+
ProtocolFeatureStore protocolFeatureStore,
50+
Message.Request request) {
4751
this.session = session;
4852
this.protocolVersion = protocolVersion;
53+
this.protocolFeatureStore = protocolFeatureStore;
4954
this.request = request;
5055
}
5156

@@ -87,21 +92,19 @@ public void onSet(
8792
table,
8893
rm.getCustomPayload().get(TabletInfo.TABLETS_ROUTING_V1_CUSTOM_PAYLOAD_KEY));
8994
}
90-
ProtocolFeatureStore featureStore =
91-
ProtocolFeatureStore.loadFromChannel(connection.channel);
9295

9396
switch (rm.kind) {
9497
case SET_KEYSPACE:
9598
// propagate the keyspace change to other connections
9699
session.poolsState.setKeyspace(((Responses.Result.SetKeyspace) rm).keyspace);
97100
set(
98101
ArrayBackedResultSet.fromMessage(
99-
rm, session, protocolVersion, info, statement, featureStore));
102+
rm, session, protocolVersion, info, statement, protocolFeatureStore));
100103
break;
101104
case SCHEMA_CHANGE:
102105
ResultSet rs =
103106
ArrayBackedResultSet.fromMessage(
104-
rm, session, protocolVersion, info, statement, featureStore);
107+
rm, session, protocolVersion, info, statement, protocolFeatureStore);
105108
final Cluster.Manager cluster = session.cluster.manager;
106109
if (!cluster.configuration.getQueryOptions().isMetadataEnabled()) {
107110
cluster.waitForSchemaAgreementAndSignal(connection, this, rs);
@@ -232,12 +235,7 @@ public void run() {
232235
default:
233236
set(
234237
ArrayBackedResultSet.fromMessage(
235-
rm,
236-
session,
237-
protocolVersion,
238-
info,
239-
statement,
240-
ProtocolFeatureStore.loadFromChannel(connection.channel)));
238+
rm, session, protocolVersion, info, statement, protocolFeatureStore));
241239
break;
242240
}
243241
break;

‎driver-core/src/main/java/com/datastax/driver/core/Message.java‎

Lines changed: 11 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -273,6 +273,12 @@ Response setWarnings(List<String> warnings) {
273273
@ChannelHandler.Sharable
274274
static class ProtocolDecoder extends MessageToMessageDecoder<Frame> {
275275

276+
private final ProtocolFeatureStore protocolFeatureStore;
277+
278+
ProtocolDecoder(ProtocolFeatureStore protocolFeatureStore) {
279+
this.protocolFeatureStore = protocolFeatureStore;
280+
}
281+
276282
@Override
277283
protected void decode(ChannelHandlerContext ctx, Frame frame, List<Object> out)
278284
throws Exception {
@@ -295,8 +301,6 @@ protected void decode(ChannelHandlerContext ctx, Frame frame, List<Object> out)
295301

296302
try {
297303
CodecRegistry codecRegistry = ctx.channel().attr(CODEC_REGISTRY_ATTRIBUTE_KEY).get();
298-
ProtocolFeatureStore protocolFeatureStore =
299-
ProtocolFeatureStore.loadFromChannel(ctx.channel());
300304
assert codecRegistry != null;
301305
Response response =
302306
Response.Type.fromOpcode(frame.header.opcode)
@@ -318,19 +322,19 @@ protected void decode(ChannelHandlerContext ctx, Frame frame, List<Object> out)
318322
static class ProtocolEncoder extends MessageToMessageEncoder<Request> {
319323

320324
final ProtocolVersion protocolVersion;
325+
final ProtocolFeatureStore protocolFeatureStore;
321326

322-
ProtocolEncoder(ProtocolVersion version) {
327+
ProtocolEncoder(ProtocolVersion version, ProtocolFeatureStore protocolFeatureStore) {
323328
this.protocolVersion = version;
329+
this.protocolFeatureStore = protocolFeatureStore;
324330
}
325331

326332
@Override
327333
protected void encode(ChannelHandlerContext ctx, Request request, List<Object> out) {
328-
ProtocolFeatureStore featureStore = ProtocolFeatureStore.loadFromChannel(ctx.channel());
329-
330334
EnumSet<Frame.Header.Flag> flags = computeFlags(request);
331-
int messageSize = encodedSize(request, featureStore);
335+
int messageSize = encodedSize(request, protocolFeatureStore);
332336
ByteBuf body = ctx.alloc().buffer(messageSize);
333-
encode(request, body, featureStore);
337+
encode(request, body, protocolFeatureStore);
334338

335339
if (body.capacity() != messageSize) {
336340
logger.debug(

‎driver-core/src/main/java/com/datastax/driver/core/ProtocolFeatureStore.java‎

Lines changed: 0 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,6 @@
1515
*/
1616
package com.datastax.driver.core;
1717

18-
import io.netty.channel.Channel;
1918
import io.netty.util.AttributeKey;
2019
import java.util.List;
2120
import java.util.Map;
@@ -69,26 +68,6 @@ public void populateStartupOptions(ProtocolVersion protocolVersion, Map<String,
6968
}
7069
}
7170

72-
/**
73-
* Stores features in both {@link Host} (for backward compatibility) and as a @{link {@link
74-
* Channel} attribute
75-
*
76-
* @param host an instance of {@link Host}
77-
* @param channel an instance of {@link Channel}
78-
*/
79-
public void storeInHostAndChannel(Host host, Channel channel) {
80-
host.setProtocolFeatureStore(this);
81-
channel.attr(ProtocolFeatureStore.CHANNEL_KEY).set(this);
82-
}
83-
84-
public static ProtocolFeatureStore loadFromChannel(Channel channel) {
85-
return channel.attr(ProtocolFeatureStore.CHANNEL_KEY).get();
86-
}
87-
88-
public static ProtocolFeatureStore loadFromHost(Host host) {
89-
return host.getProtocolFeatureStore();
90-
}
91-
9271
public ShardingInfo.ConnectionShardingInfo getConnectionShardingInfo() {
9372
return connectionShardingInfo;
9473
}

‎driver-core/src/main/java/com/datastax/driver/core/SchemaParser.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -721,7 +721,8 @@ private static ResultSetFuture queryAsync(
721721
String query, Connection connection, ProtocolVersion protocolVersion)
722722
throws ConnectionException, BusyConnectionException {
723723
DefaultResultSetFuture future =
724-
new DefaultResultSetFuture(null, protocolVersion, new Requests.Query(query));
724+
new DefaultResultSetFuture(
725+
null, protocolVersion, connection.getProtocolFeatureStore(), new Requests.Query(query));
725726
connection.write(future);
726727
return future;
727728
}

0 commit comments

Comments
 (0)