[incubator-inlong] branch master updated: [INLONG-4075][Manager] Add update, delete source, sink and transform APIs in manager client (#4076)

[go...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

gosonzhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-inlong.git


The following commit(s) were added to refs/heads/master by this push:
     new 3014854d3 [INLONG-4075][Manager] Add update, delete source, sink and transform APIs in manager client (#4076)
3014854d3 is described below

commit 3014854d360a23aa83f30e748caf26cb13973d3f
Author: kipshi <48...@users.noreply.github.com>
AuthorDate: Fri May 6 09:59:09 2022 +0800

    [INLONG-4075][Manager] Add update, delete source, sink and transform APIs in manager client (#4076)
---
 .../inlong/manager/client/api/InlongStream.java    | 12 ++++++++
 .../manager/client/api/impl/InlongStreamImpl.java  | 33 ++++++++++++++++++++++
 2 files changed, 45 insertions(+)

diff --git a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/InlongStream.java b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/InlongStream.java
index 26a3a2c6f..83e57524f 100644
--- a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/InlongStream.java
+++ b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/InlongStream.java
@@ -44,6 +44,18 @@ public abstract class InlongStream {
 
     public abstract void addTransform(StreamTransform transform);
 
+    public abstract void deleteSource(String sourceName);
+
+    public abstract void deleteSink(String sinkName);
+
+    public abstract void deleteTransform(String transformName);
+
+    public abstract void updateSource(StreamSource source);
+
+    public abstract void updateSink(StreamSink sink);
+
+    public abstract void updateTransform(StreamTransform transform);
+
     public abstract StreamPipeline createPipeline();
 
     public abstract InlongStream update();
diff --git a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/impl/InlongStreamImpl.java b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/impl/InlongStreamImpl.java
index 60515cf1f..1bbb01f9f 100644
--- a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/impl/InlongStreamImpl.java
+++ b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/impl/InlongStreamImpl.java
@@ -176,6 +176,39 @@ public class InlongStreamImpl extends InlongStream {
         streamTransforms.put(transformName, transform);
     }
 
+    @Override
+    public void deleteSource(String sourceName) {
+        streamSources.remove(sourceName);
+    }
+
+    @Override
+    public void deleteSink(String sinkName) {
+        streamSinks.remove(sinkName);
+    }
+
+    @Override
+    public void deleteTransform(String transformName) {
+        streamTransforms.remove(transformName);
+    }
+
+    @Override
+    public void updateSource(StreamSource source) {
+        AssertUtil.notNull(source.getSourceName(), "Source name should not be empty");
+        streamSources.put(source.getSourceName(), source);
+    }
+
+    @Override
+    public void updateSink(StreamSink sink) {
+        AssertUtil.notNull(sink.getSinkName(), "Sink name should not be empty");
+        streamSinks.put(sink.getSinkName(), sink);
+    }
+
+    @Override
+    public void updateTransform(StreamTransform transform) {
+        AssertUtil.notNull(transform.getTransformName(), "Transform name should not be empty");
+        streamTransforms.put(transform.getTransformName(), transform);
+    }
+
     @Override
     public StreamPipeline createPipeline() {
         StreamPipeline streamPipeline = new StreamPipeline();