This is an automated email from the ASF dual-hosted git repository.
zehnder pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/streampipes.git
The following commit(s) were added to refs/heads/dev by this push:
new 9128081ee0 add pipelineApi (#2706)
9128081ee0 is described below
commit 9128081ee0019a15e1c5cb398d26c026485293a1
Author: wyyolo <[email protected]>
AuthorDate: Wed May 8 17:54:49 2024 +0800
add pipelineApi (#2706)
---
.../internal/serializer/deserializer.go | 36 ++++++++
streampipes-client-go/streampipes/model/common.go | 12 +++
.../model/pipeline/pipeline_category.go | 25 +++++
streampipes-client-go/streampipes/pipeline_api.go | 102 +++++++++++++++++++++
.../streampipes/streampipes_client.go | 5 +
5 files changed, 180 insertions(+)
diff --git
a/streampipes-client-go/streampipes/internal/serializer/deserializer.go
b/streampipes-client-go/streampipes/internal/serializer/deserializer.go
index a8e8ad74ce..f3e7cfef0d 100644
--- a/streampipes-client-go/streampipes/internal/serializer/deserializer.go
+++ b/streampipes-client-go/streampipes/internal/serializer/deserializer.go
@@ -19,12 +19,17 @@ package serializer
import (
"encoding/json"
+ "github.com/apache/streampipes/streampipes-client-go/streampipes/model"
"github.com/apache/streampipes/streampipes-client-go/streampipes/model/data_lake"
+
"github.com/apache/streampipes/streampipes-client-go/streampipes/model/pipeline"
+
+
"github.com/apache/streampipes/streampipes-client-go/streampipes/model/functions"
"github.com/apache/streampipes/streampipes-client-go/streampipes/model/streampipes_user"
+
"github.com/apache/streampipes/streampipes-client-go/streampipes/model/streampipes_version"
)
@@ -96,6 +101,37 @@ func (d *StreamPipesVersionDeserializer) Unmarshal(data
[]byte) (interface{}, er
return dataSeries, nil
}
+
+type ResponseMessageDeserializer struct{}
+
+func NewResponseMessageDeserializer() *ResponseMessageDeserializer {
+ return &ResponseMessageDeserializer{}
+}
+
+func (r *ResponseMessageDeserializer) Unmarshal(data []byte) (interface{},
error) {
+ var responseMessage model.ResponseMessage
+ err := json.Unmarshal(data, &responseMessage)
+ if err != nil {
+ return nil, err
+ }
+ return responseMessage, nil
+}
+
+type PipelineCategoriesDeserializer struct{}
+
+func NewPipelineCategoriesDeserializer() *PipelineCategoriesDeserializer {
+ return &PipelineCategoriesDeserializer{}
+}
+
+func (p *PipelineCategoriesDeserializer) Unmarshal(data []byte) (interface{},
error) {
+ var pipelineCategory []pipeline.PipelineCategory
+ err := json.Unmarshal(data, &pipelineCategory)
+ if err != nil {
+ return nil, err
+ }
+ return pipelineCategory, nil
+}
+
type DataLakeDashboardDeserializer struct{}
func NewDataLakeDashboardDeserializer() *DataLakeDashboardDeserializer {
diff --git a/streampipes-client-go/streampipes/model/common.go
b/streampipes-client-go/streampipes/model/common.go
index f0ba11637a..cb235892eb 100644
--- a/streampipes-client-go/streampipes/model/common.go
+++ b/streampipes-client-go/streampipes/model/common.go
@@ -66,3 +66,15 @@ type DataSeries struct {
Headers []string `json:"http_headers"`
Tags map[string]string `json:"tags"`
}
+
+type ResponseMessage struct {
+ Success bool `json:"success"`
+ ElementName interface{} `json:"elementName"`
+ Notifications []Notification `json:"notifications"`
+}
+
+type Notification struct {
+ Title string `json:"title"`
+ Description string `json:"description"`
+ AdditionalInformation string `json:"additionalInformation"`
+}
diff --git
a/streampipes-client-go/streampipes/model/pipeline/pipeline_category.go
b/streampipes-client-go/streampipes/model/pipeline/pipeline_category.go
new file mode 100644
index 0000000000..9430ed8928
--- /dev/null
+++ b/streampipes-client-go/streampipes/model/pipeline/pipeline_category.go
@@ -0,0 +1,25 @@
+//
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+
+package pipeline
+
+type PipelineCategory struct {
+ CategoryName string `json:"categoryName"`
+ CategoryDescription string `json:"categoryDescription"`
+ ID string `json:"_id"`
+ Rev string `json:"_rev"`
+}
diff --git a/streampipes-client-go/streampipes/pipeline_api.go
b/streampipes-client-go/streampipes/pipeline_api.go
new file mode 100644
index 0000000000..5f3155aa3a
--- /dev/null
+++ b/streampipes-client-go/streampipes/pipeline_api.go
@@ -0,0 +1,102 @@
+//
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+
+package streampipes
+
+import (
+ "github.com/apache/streampipes/streampipes-client-go/streampipes/config"
+
"github.com/apache/streampipes/streampipes-client-go/streampipes/internal/serializer"
+
"github.com/apache/streampipes/streampipes-client-go/streampipes/internal/util"
+ "github.com/apache/streampipes/streampipes-client-go/streampipes/model"
+
"github.com/apache/streampipes/streampipes-client-go/streampipes/model/pipeline"
+ "io"
+ "log"
+ "net/http"
+)
+
+type Pipeline struct {
+ endpoint
+}
+
+func NewPipeline(clientConfig config.StreamPipesClientConfig) *Pipeline {
+
+ return &Pipeline{
+ endpoint{config: clientConfig},
+ }
+}
+
+func (p *Pipeline) GetPipelineCategory() ([]pipeline.PipelineCategory, error) {
+
+ endPointUrl := util.NewStreamPipesApiPath(p.config.Url,
"streampipes-backend/api/v2/pipelinecategories", nil)
+ log.Printf("Get data from: %s", endPointUrl)
+
+ response, err := p.executeRequest("GET", endPointUrl)
+ if err != nil {
+ return nil, err
+ }
+
+ if response.StatusCode != http.StatusOK {
+ err = p.handleStatusCode(response)
+ if err != nil {
+ return nil, err
+ }
+ }
+
+ body, err := io.ReadAll(response.Body)
+ if err != nil {
+ return nil, err
+ }
+
+ unmarshalData, err :=
serializer.NewPipelineCategoriesDeserializer().Unmarshal(body)
+ if err != nil {
+ return nil, err
+ }
+ pipelineCategory := unmarshalData.([]pipeline.PipelineCategory)
+
+ return pipelineCategory, nil
+}
+
+func (p *Pipeline) DeletePipelineCategory(categoryId string)
(model.ResponseMessage, error) {
+
+ endPointUrl := util.NewStreamPipesApiPath(p.config.Url,
"streampipes-backend/api/v2/pipelinecategories", []string{categoryId})
+ log.Printf("Delete data from: %s", endPointUrl)
+
+ response, err := p.executeRequest("DELETE", endPointUrl)
+ if err != nil {
+ return model.ResponseMessage{}, err
+ }
+
+ if response.StatusCode != http.StatusOK {
+ err = p.handleStatusCode(response)
+ if err != nil {
+ return model.ResponseMessage{}, err
+ }
+ }
+
+ body, err := io.ReadAll(response.Body)
+ if err != nil {
+ return model.ResponseMessage{}, err
+ }
+
+ unmarshalData, err :=
serializer.NewResponseMessageDeserializer().Unmarshal(body)
+ if err != nil {
+ return model.ResponseMessage{}, err
+ }
+ responseMessage := unmarshalData.(model.ResponseMessage)
+
+ return responseMessage, nil
+}
diff --git a/streampipes-client-go/streampipes/streampipes_client.go
b/streampipes-client-go/streampipes/streampipes_client.go
index a2e7b06990..237e3b14b4 100644
--- a/streampipes-client-go/streampipes/streampipes_client.go
+++ b/streampipes-client-go/streampipes/streampipes_client.go
@@ -73,6 +73,11 @@ func (s *StreamPipesClient) StreamPipesVersion() *Versions {
}
+func (s *StreamPipesClient) Pipeline() *Pipeline {
+
+ return NewPipeline(s.config)
+}
+
func (s *StreamPipesClient) DataLakeDashboard() *DataLakeDashboard {
return NewDataLakeDashborad(s.config)
