Skip to content

Commit c2b0e03

Browse files
kevinten10kevinten10
committed
feat: add metric for rpc headers
+ add more log + update cloud-runtimes-api version to 1.0.4.RELEASE
1 parent 1106811 commit c2b0e03

File tree

5 files changed

+91
-19
lines changed

5 files changed

+91
-19
lines changed

‎examples/src/main/java/group/rxcloud/capa/examples/rpc/DemoRpcClient.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,7 @@ public class DemoRpcClient {
2727
/**
2828
* Identifier in Capa for the service this client will invoke.
2929
*/
30-
private static final String SERVICE_APP_ID = "test";
30+
private static final String SERVICE_APP_ID = "12345.helloworld";
3131

3232
public static void main(String[] args) {
3333
CapaRpcClient capaRpcClient = new CapaRpcClientBuilder().build();

‎pom.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -72,7 +72,7 @@
7272
<java.version>8</java.version>
7373
<file.encoding>UTF-8</file.encoding>
7474
<maven.version>3.8.1</maven.version>
75-
<cloud-runtimes.version>1.0.1.RELEASE</cloud-runtimes.version>
75+
<cloud-runtimes.version>1.0.4.RELEASE</cloud-runtimes.version>
7676
<reactor-core.version>3.3.11.RELEASE</reactor-core.version>
7777
<junit.version>5.3.1</junit.version>
7878
<mockito-core.version>3.6.0</mockito-core.version>

‎sdk-spi/src/main/java/group/rxcloud/capa/spi/http/CapaHttpSpi.java‎

Lines changed: 42 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@
2222
import group.rxcloud.capa.spi.config.CapaSpiOptionsLoader;
2323
import group.rxcloud.capa.spi.config.CapaSpiProperties;
2424
import group.rxcloud.capa.spi.config.RpcServiceOptions;
25+
import group.rxcloud.cloudruntimes.domain.core.invocation.HttpExtension;
2526
import group.rxcloud.cloudruntimes.utils.TypeRef;
2627
import okhttp3.OkHttpClient;
2728
import org.slf4j.Logger;
@@ -47,8 +48,8 @@ public CapaHttpSpi(OkHttpClient httpClient, CapaObjectSerializer objectSerialize
4748
/**
4849
* Templates, delegate to specific http invoker.
4950
*
50-
* @param httpMethod Ignore, fix to POST. TODO
51-
* @param urlParameters Ignore, fix to EMPTY. TODO
51+
* @param httpMethod Ignore, fix to POST. FIXME
52+
* @param urlParameters Ignore, fix to EMPTY. FIXME
5253
*/
5354
@Override
5455
protected <T> CompletableFuture<HttpResponse<T>> doInvokeApi(String httpMethod,
@@ -72,6 +73,22 @@ protected <T> CompletableFuture<HttpResponse<T>> doInvokeApi(String httpMethod,
7273
logger.debug("[CapaHttpSpi] invoke rpc context[{}]", context);
7374
}
7475
}
76+
// FIXME Ignore, fix to POST.
77+
if (!HttpExtension.POST.getMethod().toString().equalsIgnoreCase(httpMethod)) {
78+
if (logger.isWarnEnabled()) {
79+
logger.warn("[CapaHttpSpi] invoke rpc httpMethod[{}] only support POST now.",
80+
httpMethod);
81+
}
82+
httpMethod = HttpExtension.POST.getMethod().toString();
83+
}
84+
// FIXME Ignore, fix to EMPTY.
85+
if (urlParameters != null && !urlParameters.isEmpty()) {
86+
if (logger.isWarnEnabled()) {
87+
logger.warn("[CapaHttpSpi] invoke rpc urlParameters[{}] not supported now.",
88+
urlParameters);
89+
}
90+
urlParameters = null;
91+
}
7592

7693
// parse url path segments
7794
Objects.requireNonNull(pathSegments, "pathSegments");
@@ -96,6 +113,29 @@ protected <T> CompletableFuture<HttpResponse<T>> doInvokeApi(String httpMethod,
96113
// spi invoke
97114
CompletableFuture<HttpResponse<T>> invokeSpiApi =
98115
invokeSpiApi(appId, method, requestData, headers, type, rpcServiceOptions);
116+
invokeSpiApi.whenComplete((tHttpResponse, throwable) -> {
117+
if (throwable != null) {
118+
if (logger.isWarnEnabled()) {
119+
logger.warn("[CapaHttpSpi] invoke rpc response error",
120+
throwable);
121+
}
122+
return;
123+
}
124+
if (tHttpResponse == null) {
125+
if (logger.isWarnEnabled()) {
126+
logger.warn("[CapaHttpSpi] invoke rpc response empty[{}]",
127+
tHttpResponse);
128+
}
129+
return;
130+
}
131+
final int responseStatusCode = tHttpResponse.getStatusCode();
132+
final Map<String, String> responseHeaders = tHttpResponse.getHeaders();
133+
final T responseBody = tHttpResponse.getBody();
134+
if (logger.isDebugEnabled()) {
135+
logger.debug("[CapaHttpSpi] invoke rpc response code[{}] headers[{}] body[{}]",
136+
responseStatusCode, responseHeaders, responseBody);
137+
}
138+
});
99139
return invokeSpiApi;
100140
}
101141

‎sdk-spi/src/main/java/group/rxcloud/capa/spi/http/CapaSerializeHttpSpi.java‎

Lines changed: 44 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -24,11 +24,12 @@
2424
import group.rxcloud.cloudruntimes.domain.core.invocation.Metadata;
2525
import group.rxcloud.cloudruntimes.utils.TypeRef;
2626
import okhttp3.Call;
27-
import okhttp3.OkHttpClient;
2827
import okhttp3.Callback;
28+
import okhttp3.Headers;
29+
import okhttp3.MediaType;
30+
import okhttp3.OkHttpClient;
2931
import okhttp3.Request;
3032
import okhttp3.RequestBody;
31-
import okhttp3.MediaType;
3233
import okhttp3.Response;
3334
import okhttp3.ResponseBody;
3435
import org.slf4j.Logger;
@@ -42,7 +43,7 @@
4243
import java.util.concurrent.CompletableFuture;
4344

4445
/**
45-
* The type Capa serialize http spi.
46+
* The Capa http spi with default serializer process.
4647
*/
4748
public abstract class CapaSerializeHttpSpi extends CapaHttpSpi {
4849

@@ -68,9 +69,17 @@ protected byte[] getRequestWithSerialize(Object requestData) {
6869
try {
6970
return objectSerializer.serialize(requestData);
7071
} catch (IOException e) {
72+
if (logger.isWarnEnabled()) {
73+
logger.warn("[CapaSerializeHttpSpi] serialize rpc request[{}] io error",
74+
requestData, e);
75+
}
7176
throw new CapaException(CapaErrorContext.PARAMETER_RPC_REQUEST_SERIALIZE_ERROR,
7277
"Request Type: " + requestData.getClass().getName());
7378
} catch (Exception e) {
79+
if (logger.isWarnEnabled()) {
80+
logger.warn("[CapaSerializeHttpSpi] serialize rpc request[{}] error",
81+
requestData, e);
82+
}
7483
throw new CapaException(CapaErrorContext.PARAMETER_RPC_REQUEST_SERIALIZE_ERROR,
7584
"Request Type: " + requestData.getClass().getName(), e);
7685
}
@@ -84,7 +93,6 @@ protected byte[] getRequestWithSerialize(Object requestData) {
8493
* @return the request body with byte[] serialize
8594
*/
8695
protected RequestBody getRequestBodyWithSerialize(Object requestData, Map<String, String> headers) {
87-
byte[] serializedRequestBody = getRequestWithSerialize(requestData);
8896
final String contentType = headers != null
8997
? headers.get(Metadata.CONTENT_TYPE)
9098
: null;
@@ -97,11 +105,27 @@ protected RequestBody getRequestBodyWithSerialize(Object requestData, Map<String
97105
? REQUEST_BODY_EMPTY_JSON
98106
: RequestBody.Companion.create(new byte[0], mediaType);
99107
} else {
108+
byte[] serializedRequestBody = getRequestWithSerialize(requestData);
100109
body = RequestBody.Companion.create(serializedRequestBody, mediaType);
101110
}
102111
return body;
103112
}
104113

114+
/**
115+
* Gets request headers with given params.
116+
*
117+
* @param headersParams user given params
118+
* @return the request headers
119+
*/
120+
protected Headers getRequestHeaderWithParams(Map<String, String> headersParams) {
121+
okhttp3.Headers.Builder headersBuilder = new okhttp3.Headers.Builder();
122+
if (headersParams == null || headersParams.size() == 0) {
123+
return headersBuilder.build();
124+
}
125+
headersParams.forEach(headersBuilder::add);
126+
return headersBuilder.build();
127+
}
128+
105129
/**
106130
* Http async call
107131
*/
@@ -130,21 +154,23 @@ protected <T> CompletableFuture<HttpResponse<T>> doAsyncInvoke0(Request request,
130154
* @return the response body with byte[] deserialize
131155
*/
132156
protected <T> HttpResponse<T> getResponseBodyWithDeserialize(TypeRef<T> type, HttpResponse<byte[]> httpResponse) {
157+
final int httpResponseStatusCode = httpResponse.getStatusCode();
158+
final Map<String, String> httpResponseHeaders = httpResponse.getHeaders();
133159
final byte[] httpResponseBody = httpResponse.getBody();
134160
try {
135161
T responseObject = objectSerializer.deserialize(httpResponseBody, type);
136-
return new HttpResponse<>(responseObject, httpResponse.getHeaders(), httpResponse.getStatusCode());
162+
return new HttpResponse<>(responseObject, httpResponseHeaders, httpResponseStatusCode);
137163
} catch (IOException e) {
138164
if (logger.isWarnEnabled()) {
139-
logger.warn("[CapaSerializeHttpSpi] deserialize rpc response[{}] type[{}] io error",
140-
httpResponseBody, type, e);
165+
logger.warn("[CapaSerializeHttpSpi] deserialize rpc statusCode[{}] headers[{}] response[{}] type[{}] io error",
166+
httpResponseStatusCode, httpResponseHeaders, httpResponseBody, type, e);
141167
}
142168
throw new CapaException(CapaErrorContext.PARAMETER_RPC_RESPONSE_DESERIALIZE_ERROR,
143169
"Response Type: " + type, e);
144170
} catch (Exception e) {
145171
if (logger.isWarnEnabled()) {
146-
logger.warn("[CapaSerializeHttpSpi] deserialize rpc response[{}] type[{}] error",
147-
httpResponseBody, type, e);
172+
logger.warn("[CapaSerializeHttpSpi] deserialize rpc statusCode[{}] headers[{}] response[{}] type[{}] error",
173+
httpResponseStatusCode, httpResponseHeaders, httpResponseBody, type, e);
148174
}
149175
throw new CapaException(CapaErrorContext.PARAMETER_RPC_RESPONSE_DESERIALIZE_ERROR,
150176
"Response Type: " + type, e);
@@ -190,9 +216,15 @@ public void onResponse(Call call, Response response) throws IOException {
190216
return;
191217
}
192218

193-
Map<String, String> mapHeaders = new HashMap<>();
194-
// response.headers()
195-
// .forEach(pair -> mapHeaders.put(pair.getFirst(), pair.getSecond()));
219+
Map<String, String> mapHeaders;
220+
Headers responseHeaders = response.headers();
221+
if (responseHeaders == null || responseHeaders.size() == 0) {
222+
mapHeaders = new HashMap<>(2, 1);
223+
} else {
224+
mapHeaders = new HashMap<>(responseHeaders.size() << 1);
225+
responseHeaders.forEach(pair -> mapHeaders.put(pair.getFirst(), pair.getSecond()));
226+
}
227+
196228
HttpResponse<byte[]> httpResponse = new HttpResponse<>(bodyBytes, mapHeaders, response.code());
197229
future.complete(httpResponse);
198230
}

‎sdk/src/main/java/group/rxcloud/capa/rpc/CapaRpcClientHttp.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -89,9 +89,9 @@ public <T> Mono<T> invokeMethod(InvokeMethodRequest invokeMethodRequest, TypeRef
8989
context,
9090
type))
9191
.flatMap(httpResponse -> {
92-
T object = httpResponse.getBody();
93-
if (object != null) {
94-
return Mono.just(object);
92+
final T responseBody = httpResponse.getBody();
93+
if (responseBody != null) {
94+
return Mono.just(responseBody);
9595
}
9696
return Mono.empty();
9797
});

0 commit comments

Comments
 (0)