1010
1111import java .io .IOException ;
1212import java .io .InputStream ;
13+ import java .io .UncheckedIOException ;
1314import java .util .List ;
1415import java .util .Map ;
1516import java .util .Map .Entry ;
1617import java .util .concurrent .CompletableFuture ;
18+ import java .util .function .Predicate ;
1719import java .util .stream .Collectors ;
1820import javax .annotation .Nullable ;
1921import org .opensearch .client .ApiClient ;
2426 * Client for the generic HTTP requests.
2527 */
2628public class OpenSearchGenericClient extends ApiClient <OpenSearchTransport , OpenSearchGenericClient > {
29+ /**
30+ * Generic client options
31+ */
32+ public static final class ClientOptions {
33+ private static final ClientOptions DEFAULT = new ClientOptions ();
34+
35+ private final Predicate <Integer > error ;
36+
37+ private ClientOptions () {
38+ this (statusCode -> false );
39+ }
40+
41+ private ClientOptions (final Predicate <Integer > error ) {
42+ this .error = error ;
43+ }
44+
45+ public static ClientOptions throwOnHttpErrors () {
46+ return new ClientOptions (statusCode -> statusCode >= 400 );
47+ }
48+ }
49+
2750 /**
2851 * Generic endpoint instance
2952 */
3053 private static final class GenericEndpoint implements org .opensearch .client .transport .GenericEndpoint <Request , Response > {
3154 private final Request request ;
55+ private final Predicate <Integer > error ;
3256
33- public GenericEndpoint (Request request ) {
57+ public GenericEndpoint (Request request , Predicate < Integer > error ) {
3458 this .request = request ;
59+ this .error = error ;
3560 }
3661
3762 @ Override
@@ -67,24 +92,70 @@ public GenericResponse responseDeserializer(
6792 int status ,
6893 String reason ,
6994 List <Entry <String , String >> headers ,
70- String contentType ,
71- InputStream body
95+ @ Nullable String contentType ,
96+ @ Nullable InputStream body
7297 ) {
73- return new GenericResponse (uri , protocol , method , status , reason , headers , Body .from (body , contentType ));
98+ if (isError (status )) {
99+ // Fully consume the response body since the it will be propagated as an exception with possible no chance to be closed
100+ try (Body b = Body .from (body , contentType )) {
101+ if (b != null ) {
102+ return new GenericResponse (
103+ uri ,
104+ protocol ,
105+ method ,
106+ status ,
107+ reason ,
108+ headers ,
109+ Body .from (b .bodyAsBytes (), b .contentType ())
110+ );
111+ } else {
112+ return new GenericResponse (uri , protocol , method , status , reason , headers );
113+ }
114+ } catch (final IOException ex ) {
115+ throw new UncheckedIOException (ex );
116+ }
117+ } else {
118+ return new GenericResponse (uri , protocol , method , status , reason , headers , Body .from (body , contentType ));
119+ }
120+ }
121+
122+ @ Override
123+ public boolean isError (int statusCode ) {
124+ return error .test (statusCode );
125+ }
126+
127+ @ Override
128+ public <T extends RuntimeException > T exceptionConverter (int statusCode , @ Nullable Response error ) {
129+ throw new OpenSearchClientException (error );
74130 }
75131 }
76132
133+ private final ClientOptions clientOptions ;
134+
77135 public OpenSearchGenericClient (OpenSearchTransport transport ) {
78- super (transport , null );
136+ this (transport , null , ClientOptions . DEFAULT );
79137 }
80138
81139 public OpenSearchGenericClient (OpenSearchTransport transport , @ Nullable TransportOptions transportOptions ) {
140+ this (transport , transportOptions , ClientOptions .DEFAULT );
141+ }
142+
143+ public OpenSearchGenericClient (
144+ OpenSearchTransport transport ,
145+ @ Nullable TransportOptions transportOptions ,
146+ ClientOptions clientOptions
147+ ) {
82148 super (transport , transportOptions );
149+ this .clientOptions = clientOptions ;
150+ }
151+
152+ public OpenSearchGenericClient withClientOptions (ClientOptions clientOptions ) {
153+ return new OpenSearchGenericClient (this .transport , this .transportOptions , clientOptions );
83154 }
84155
85156 @ Override
86157 public OpenSearchGenericClient withTransportOptions (@ Nullable TransportOptions transportOptions ) {
87- return new OpenSearchGenericClient (this .transport , transportOptions );
158+ return new OpenSearchGenericClient (this .transport , transportOptions , this . clientOptions );
88159 }
89160
90161 /**
@@ -94,7 +165,7 @@ public OpenSearchGenericClient withTransportOptions(@Nullable TransportOptions t
94165 * @throws IOException I/O exception
95166 */
96167 public Response execute (Request request ) throws IOException {
97- return transport .performRequest (request , new GenericEndpoint (request ), this .transportOptions );
168+ return transport .performRequest (request , new GenericEndpoint (request , clientOptions . error ), this .transportOptions );
98169 }
99170
100171 /**
@@ -103,6 +174,6 @@ public Response execute(Request request) throws IOException {
103174 * @return generic HTTP response future
104175 */
105176 public CompletableFuture <Response > executeAsync (Request request ) {
106- return transport .performRequestAsync (request , new GenericEndpoint (request ), this .transportOptions );
177+ return transport .performRequestAsync (request , new GenericEndpoint (request , clientOptions . error ), this .transportOptions );
107178 }
108179}
![opensearch-trigger-bot[bot]](https://avatars.githubusercontent.com/u/80134844?v=4&size=40){kind=avatar}
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?v=4&size=40){kind=avatar}
0 commit comments