Skip to content
Permalink
Browse files
[INLONG-4203][Manager] Improve HTTP request and response parse in man…
…ager client (#4204)
  • Loading branch information
leosanqing committed May 16, 2022
1 parent 2ec59f7 commit 0553b321e560659a6b4393b33459b029500ea9aa
Showing 3 changed files with 134 additions and 151 deletions.
@@ -121,25 +121,22 @@ public InlongStream init() {
InlongStreamInfo streamInfo = streamContext.getStreamInfo();
StreamPipeline streamPipeline = inlongStream.createPipeline();
streamInfo.setTempView(GsonUtil.toJson(streamPipeline));
String streamIndex = managerClient.createStreamInfo(streamInfo);
streamInfo.setId(Double.valueOf(streamIndex).intValue());
Double streamIndex = managerClient.createStreamInfo(streamInfo);
streamInfo.setId(streamIndex.intValue());
//Create source and update index
List<SourceRequest> sourceRequests = Lists.newArrayList(streamContext.getSourceRequests().values());
for (SourceRequest sourceRequest : sourceRequests) {
String sourceIndex = managerClient.createSource(sourceRequest);
sourceRequest.setId(Double.valueOf(sourceIndex).intValue());
sourceRequest.setId(managerClient.createSource(sourceRequest).intValue());
}
//Create sink and update index
List<SinkRequest> sinkRequests = Lists.newArrayList(streamContext.getSinkRequests().values());
for (SinkRequest sinkRequest : sinkRequests) {
String sinkIndex = managerClient.createSink(sinkRequest);
sinkRequest.setId(Double.valueOf(sinkIndex).intValue());
sinkRequest.setId(managerClient.createSink(sinkRequest).intValue());
}
//Create transform and update index
List<TransformRequest> transformRequests = Lists.newArrayList(streamContext.getTransformRequests().values());
for (TransformRequest transformRequest : transformRequests) {
String transformIndex = managerClient.createTransform(transformRequest);
transformRequest.setId(Double.valueOf(transformIndex).intValue());
transformRequest.setId(managerClient.createTransform(transformRequest).intValue());
}
return inlongStream;
}
@@ -200,8 +197,7 @@ private void initOrUpdateTransform() {
continue;
}
TransformRequest transformRequest = requestEntry.getValue();
String index = managerClient.createTransform(transformRequest);
transformRequest.setId(Double.valueOf(index).intValue());
transformRequest.setId(managerClient.createTransform(transformRequest).intValue());
}
}

@@ -238,8 +234,7 @@ private void initOrUpdateSource() {
continue;
}
SourceRequest sourceRequest = requestEntry.getValue();
String index = managerClient.createSource(sourceRequest);
sourceRequest.setId(Double.valueOf(index).intValue());
sourceRequest.setId(managerClient.createSource(sourceRequest).intValue());
}
}

@@ -276,8 +271,7 @@ private void initOrUpdateSink() {
continue;
}
SinkRequest sinkRequest = requestEntry.getValue();
String index = managerClient.createSink(sinkRequest);
sinkRequest.setId(Double.valueOf(index).intValue());
sinkRequest.setId(managerClient.createSink(sinkRequest).intValue());
}
}
}

0 comments on commit 0553b32

Please sign in to comment.