3838 */
3939public final class ConnectionProfile {
4040
41- /**
42- * Builds a connection profile that is dedicated to a single channel type. Use this
43- * when opening single use connections
44- */
45- public static ConnectionProfile buildSingleChannelProfile (TransportRequestOptions .Type channelType ,
46- @ Nullable TimeValue connectTimeout ,
47- @ Nullable TimeValue handshakeTimeout ) {
48- Builder builder = new Builder ();
49- builder .addConnections (1 , channelType );
50- final EnumSet <TransportRequestOptions .Type > otherTypes = EnumSet .allOf (TransportRequestOptions .Type .class );
51- otherTypes .remove (channelType );
52- builder .addConnections (0 , otherTypes .stream ().toArray (TransportRequestOptions .Type []::new ));
53- if (connectTimeout != null ) {
54- builder .setConnectTimeout (connectTimeout );
55- }
56- if (handshakeTimeout != null ) {
57- builder .setHandshakeTimeout (handshakeTimeout );
58- }
59- return builder .build ();
60- }
61-
62- private final List <ConnectionTypeHandle > handles ;
63- private final int numConnections ;
64- private final TimeValue connectTimeout ;
65- private final TimeValue handshakeTimeout ;
66-
67- private ConnectionProfile (List <ConnectionTypeHandle > handles , int numConnections , TimeValue connectTimeout ,
68- TimeValue handshakeTimeout ) {
69- this .handles = handles ;
70- this .numConnections = numConnections ;
71- this .connectTimeout = connectTimeout ;
72- this .handshakeTimeout = handshakeTimeout ;
73- }
74-
7541 /**
7642 * takes a {@link ConnectionProfile} resolves it to a fully specified (i.e., no nulls) profile
7743 */
7844 public static ConnectionProfile resolveConnectionProfile (@ Nullable ConnectionProfile profile , ConnectionProfile fallbackProfile ) {
7945 Objects .requireNonNull (fallbackProfile );
8046 if (profile == null ) {
8147 return fallbackProfile ;
82- } else if (profile .getConnectTimeout () != null && profile .getHandshakeTimeout () != null ) {
48+ } else if (profile .getConnectTimeout () != null && profile .getHandshakeTimeout () != null
49+ && profile .getCompressionEnabled () != null ) {
8350 return profile ;
8451 } else {
8552 ConnectionProfile .Builder builder = new ConnectionProfile .Builder (profile );
@@ -89,6 +56,9 @@ public static ConnectionProfile resolveConnectionProfile(@Nullable ConnectionPro
8956 if (profile .getHandshakeTimeout () == null ) {
9057 builder .setHandshakeTimeout (fallbackProfile .getHandshakeTimeout ());
9158 }
59+ if (profile .getCompressionEnabled () == null ) {
60+ builder .setCompressionEnabled (fallbackProfile .getCompressionEnabled ());
61+ }
9262 return builder .build ();
9363 }
9464 }
@@ -108,6 +78,7 @@ public static ConnectionProfile buildDefaultConnectionProfile(Settings settings)
10878 Builder builder = new Builder ();
10979 builder .setConnectTimeout (TransportService .TCP_CONNECT_TIMEOUT .get (settings ));
11080 builder .setHandshakeTimeout (TransportService .TCP_CONNECT_TIMEOUT .get (settings ));
81+ builder .setCompressionEnabled (Transport .TRANSPORT_TCP_COMPRESS .get (settings ));
11182 builder .addConnections (connectionsPerNodeBulk , TransportRequestOptions .Type .BULK );
11283 builder .addConnections (connectionsPerNodePing , TransportRequestOptions .Type .PING );
11384 // if we are not master eligible we don't need a dedicated channel to publish the state
@@ -118,13 +89,77 @@ public static ConnectionProfile buildDefaultConnectionProfile(Settings settings)
11889 return builder .build ();
11990 }
12091
92+ /**
93+ * Builds a connection profile that is dedicated to a single channel type. Use this
94+ * when opening single use connections
95+ */
96+ public static ConnectionProfile buildSingleChannelProfile (TransportRequestOptions .Type channelType ) {
97+ return buildSingleChannelProfile (channelType , null , null , null );
98+ }
99+
100+ /**
101+ * Builds a connection profile that is dedicated to a single channel type. Allows passing compression
102+ * settings.
103+ */
104+ public static ConnectionProfile buildSingleChannelProfile (TransportRequestOptions .Type channelType , boolean compressionEnabled ) {
105+ return buildSingleChannelProfile (channelType , null , null , compressionEnabled );
106+ }
107+
108+ /**
109+ * Builds a connection profile that is dedicated to a single channel type. Allows passing connection and
110+ * handshake timeouts.
111+ */
112+ public static ConnectionProfile buildSingleChannelProfile (TransportRequestOptions .Type channelType , @ Nullable TimeValue connectTimeout ,
113+ @ Nullable TimeValue handshakeTimeout ) {
114+ return buildSingleChannelProfile (channelType , connectTimeout , handshakeTimeout , null );
115+ }
116+
117+ /**
118+ * Builds a connection profile that is dedicated to a single channel type. Allows passing connection and
119+ * handshake timeouts and compression settings.
120+ */
121+ public static ConnectionProfile buildSingleChannelProfile (TransportRequestOptions .Type channelType , @ Nullable TimeValue connectTimeout ,
122+ @ Nullable TimeValue handshakeTimeout , @ Nullable Boolean compressionEnabled ) {
123+ Builder builder = new Builder ();
124+ builder .addConnections (1 , channelType );
125+ final EnumSet <TransportRequestOptions .Type > otherTypes = EnumSet .allOf (TransportRequestOptions .Type .class );
126+ otherTypes .remove (channelType );
127+ builder .addConnections (0 , otherTypes .toArray (new TransportRequestOptions .Type [0 ]));
128+ if (connectTimeout != null ) {
129+ builder .setConnectTimeout (connectTimeout );
130+ }
131+ if (handshakeTimeout != null ) {
132+ builder .setHandshakeTimeout (handshakeTimeout );
133+ }
134+ if (compressionEnabled != null ) {
135+ builder .setCompressionEnabled (compressionEnabled );
136+ }
137+ return builder .build ();
138+ }
139+
140+ private final List <ConnectionTypeHandle > handles ;
141+ private final int numConnections ;
142+ private final TimeValue connectTimeout ;
143+ private final TimeValue handshakeTimeout ;
144+ private final Boolean compressionEnabled ;
145+
146+ private ConnectionProfile (List <ConnectionTypeHandle > handles , int numConnections , TimeValue connectTimeout ,
147+ TimeValue handshakeTimeout , Boolean compressionEnabled ) {
148+ this .handles = handles ;
149+ this .numConnections = numConnections ;
150+ this .connectTimeout = connectTimeout ;
151+ this .handshakeTimeout = handshakeTimeout ;
152+ this .compressionEnabled = compressionEnabled ;
153+ }
154+
121155 /**
122156 * A builder to build a new {@link ConnectionProfile}
123157 */
124158 public static class Builder {
125159 private final List <ConnectionTypeHandle > handles = new ArrayList <>();
126160 private final Set <TransportRequestOptions .Type > addedTypes = EnumSet .noneOf (TransportRequestOptions .Type .class );
127- private int offset = 0 ;
161+ private int numConnections = 0 ;
162+ private Boolean compressionEnabled ;
128163 private TimeValue connectTimeout ;
129164 private TimeValue handshakeTimeout ;
130165
@@ -135,10 +170,11 @@ public Builder() {
135170 /** copy constructor, using another profile as a base */
136171 public Builder (ConnectionProfile source ) {
137172 handles .addAll (source .getHandles ());
138- offset = source .getNumConnections ();
173+ numConnections = source .getNumConnections ();
139174 handles .forEach (th -> addedTypes .addAll (th .types ));
140175 connectTimeout = source .getConnectTimeout ();
141176 handshakeTimeout = source .getHandshakeTimeout ();
177+ compressionEnabled = source .getCompressionEnabled ();
142178 }
143179 /**
144180 * Sets a connect timeout for this connection profile
@@ -160,6 +196,13 @@ public void setHandshakeTimeout(TimeValue handshakeTimeout) {
160196 this .handshakeTimeout = handshakeTimeout ;
161197 }
162198
199+ /**
200+ * Sets compression enabled for this connection profile
201+ */
202+ public void setCompressionEnabled (boolean compressionEnabled ) {
203+ this .compressionEnabled = compressionEnabled ;
204+ }
205+
163206 /**
164207 * Adds a number of connections for one or more types. Each type can only be added once.
165208 * @param numConnections the number of connections to use in the pool for the given connection types
@@ -175,8 +218,8 @@ public void addConnections(int numConnections, TransportRequestOptions.Type... t
175218 }
176219 }
177220 addedTypes .addAll (Arrays .asList (types ));
178- handles .add (new ConnectionTypeHandle (offset , numConnections , EnumSet .copyOf (Arrays .asList (types ))));
179- offset += numConnections ;
221+ handles .add (new ConnectionTypeHandle (this . numConnections , numConnections , EnumSet .copyOf (Arrays .asList (types ))));
222+ this . numConnections += numConnections ;
180223 }
181224
182225 /**
@@ -189,7 +232,8 @@ public ConnectionProfile build() {
189232 if (types .isEmpty () == false ) {
190233 throw new IllegalStateException ("not all types are added for this connection profile - missing types: " + types );
191234 }
192- return new ConnectionProfile (Collections .unmodifiableList (handles ), offset , connectTimeout , handshakeTimeout );
235+ return new ConnectionProfile (Collections .unmodifiableList (handles ), numConnections , connectTimeout , handshakeTimeout ,
236+ compressionEnabled );
193237 }
194238
195239 }
@@ -208,6 +252,14 @@ public TimeValue getHandshakeTimeout() {
208252 return handshakeTimeout ;
209253 }
210254
255+ /**
256+ * Returns boolean indicating if compression is enabled or <code>null</code> if no explicit compression
257+ * is set on this profile.
258+ */
259+ public Boolean getCompressionEnabled () {
260+ return compressionEnabled ;
261+ }
262+
211263 /**
212264 * Returns the total number of connections for this profile
213265 */
0 commit comments