feat: add more support for pipeline API to Go client

apache · Jun 12, 2024 · b7718a5 · b7718a5
1 parent 2bd36aa
commit b7718a5
Show file tree

Hide file tree

Showing 11 changed files with 496 additions and 37 deletions.
diff --git a/streampipes-client-go/streampipes/data_lake_dashboard_api.go b/streampipes-client-go/streampipes/data_lake_dashboard_api.go
@@ -42,7 +42,7 @@ func (d *DataLakeDashboard) GetSingleDataLakeDashboard(dashboardId string) (data
 	endPointUrl := util.NewStreamPipesApiPath(d.config.Url, "streampipes-backend/api/v3/datalake/dashboard", []string{dashboardId})
 	log.Printf("Get data from: %s", endPointUrl)
 
-	response, err := d.executeRequest("GET", endPointUrl)
+	response, err := d.executeRequest("GET", endPointUrl, nil)
 	if err != nil {
 		return data_lake.Dashboard{}, err
 	}
@@ -72,7 +72,7 @@ func (d *DataLakeDashboard) GetAllDataLakeDashboard() ([]data_lake.Dashboard, er
 	endPointUrl := util.NewStreamPipesApiPath(d.config.Url, "streampipes-backend/api/v3/datalake/dashboard", nil)
 	log.Printf("Get data from: %s", endPointUrl)
 
-	response, err := d.executeRequest("GET", endPointUrl)
+	response, err := d.executeRequest("GET", endPointUrl, nil)
 	if err != nil {
 		return nil, err
 	}
@@ -102,7 +102,7 @@ func (d *DataLakeDashboard) DeleteSingleDataLakeDashboard(dashboardId string) er
 	endPointUrl := util.NewStreamPipesApiPath(d.config.Url, "streampipes-backend/api/v3/datalake/dashboard", []string{dashboardId})
 	log.Printf("Delete data from: %s", endPointUrl)
 
-	response, err := d.executeRequest("DELETE", endPointUrl)
+	response, err := d.executeRequest("DELETE", endPointUrl, nil)
 	if err != nil {
 		return err
 	}

diff --git a/streampipes-client-go/streampipes/data_lake_widget_api.go b/streampipes-client-go/streampipes/data_lake_widget_api.go
@@ -43,7 +43,7 @@ func (d *DataLakeWidget) GetSingleDataLakeWidget(widgetId string) (data_lake.Dat
 	endPointUrl := util.NewStreamPipesApiPath(d.config.Url, "streampipes-backend/api/v3/datalake/dashboard/widgets", []string{widgetId})
 	log.Printf("Get data from: %s", endPointUrl)
 
-	response, err := d.executeRequest("GET", endPointUrl)
+	response, err := d.executeRequest("GET", endPointUrl, nil)
 	if err != nil {
 		return data_lake.DataExplorerWidgetModel{}, err
 	}
@@ -74,7 +74,7 @@ func (d *DataLakeWidget) GetAllDataLakeWidget() ([]data_lake.DataExplorerWidgetM
 	endPointUrl := util.NewStreamPipesApiPath(d.config.Url, "streampipes-backend/api/v3/datalake/dashboard/widgets", nil)
 	log.Printf("Get data from: %s", endPointUrl)
 
-	response, err := d.executeRequest("GET", endPointUrl)
+	response, err := d.executeRequest("GET", endPointUrl, nil)
 	if err != nil {
 		return nil, err
 	}
@@ -104,7 +104,7 @@ func (d *DataLakeWidget) DeleteSingleDataLakeWidget(widgetId string) error {
 	endPointUrl := util.NewStreamPipesApiPath(d.config.Url, "streampipes-backend/api/v3/datalake/dashboard/widgets", []string{widgetId})
 	log.Printf("Delete data from: %s", endPointUrl)
 
-	response, err := d.executeRequest("DELETE", endPointUrl)
+	response, err := d.executeRequest("DELETE", endPointUrl, nil)
 	if err != nil {
 		return err
 	}

diff --git a/streampipes-client-go/streampipes/endpoint.go b/streampipes-client-go/streampipes/endpoint.go
@@ -18,19 +18,24 @@
 package streampipes
 
 import (
+	"bytes"
 	"errors"
 	"github.com/apache/streampipes/streampipes-client-go/streampipes/config"
 	headers "github.com/apache/streampipes/streampipes-client-go/streampipes/internal/http_headers"
+	"io"
 	"net/http"
 )
 
 type endpoint struct {
 	config config.StreamPipesClientConfig
 }
 
-func (e *endpoint) executeRequest(method string, endPointUrl string) (*http.Response, error) {
-
-	req, err := http.NewRequest(method, endPointUrl, nil)
+func (e *endpoint) executeRequest(method string, endPointUrl string, body []byte) (*http.Response, error) {
+	var reader io.Reader
+	if body != nil {
+		reader = bytes.NewReader(body)
+	}
+	req, err := http.NewRequest(method, endPointUrl, reader)
 	if err != nil {
 		return nil, err
 	}

diff --git a/streampipes-client-go/streampipes/functions_api.go b/streampipes-client-go/streampipes/functions_api.go
@@ -43,7 +43,7 @@ func (f *Functions) GetFunctionLogs(functionId string) ([]functions.SpLogEntry,
 	endPointUrl := util.NewStreamPipesApiPath(f.config.Url, "streampipes-backend/api/v2/functions", []string{functionId, "logs"})
 	log.Printf("Get data from: %s", endPointUrl)
 
-	response, err := f.executeRequest("GET", endPointUrl)
+	response, err := f.executeRequest("GET", endPointUrl, nil)
 	if err != nil {
 		return nil, err
 	}
@@ -74,7 +74,7 @@ func (f *Functions) GetFunctionMetrics(functionId string) (functions.SpMetricsEn
 	endPointUrl := util.NewStreamPipesApiPath(f.config.Url, "streampipes-backend/api/v2/functions", []string{functionId, "metrics"})
 	log.Printf("Get data from: %s", endPointUrl)
 
-	response, err := f.executeRequest("GET", endPointUrl)
+	response, err := f.executeRequest("GET", endPointUrl, nil)
 	if err != nil {
 		return functions.SpMetricsEntry{}, err
 	}
@@ -105,7 +105,7 @@ func (f *Functions) GetAllFunction() ([]functions.FunctionDefinition, error) {
 	endPointUrl := util.NewStreamPipesApiPath(f.config.Url, "streampipes-backend/api/v2/functions", nil)
 	log.Printf("Get data from: %s", endPointUrl)
 
-	response, err := f.executeRequest("GET", endPointUrl)
+	response, err := f.executeRequest("GET", endPointUrl, nil)
 	if err != nil {
 		return nil, err
 	}
@@ -135,7 +135,7 @@ func (f *Functions) DeleteSingleFunction(functionId string) error {
 	endPointUrl := util.NewStreamPipesApiPath(f.config.Url, "streampipes-backend/api/v2/functions", []string{functionId})
 	log.Printf("Delete data from: %s", endPointUrl)
 
-	response, err := f.executeRequest("DELETE", endPointUrl)
+	response, err := f.executeRequest("DELETE", endPointUrl, nil)
 	if err != nil {
 		return err
 	}

diff --git a/streampipes-client-go/streampipes/internal/serializer/deserializer.go b/streampipes-client-go/streampipes/internal/serializer/deserializer.go
@@ -24,12 +24,10 @@ import (
 
 	"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"
 )
 
@@ -101,7 +99,6 @@ func (d *StreamPipesVersionDeserializer) Unmarshal(data []byte) (interface{}, er
 	return dataSeries, nil
 }
 
-
 type ResponseMessageDeserializer struct{}
 
 func NewResponseMessageDeserializer() *ResponseMessageDeserializer {
@@ -159,7 +156,7 @@ func (d *DataLakeDashboardsDeserializer) Unmarshal(data []byte) (interface{}, er
 	if err != nil {
 		return nil, err
 	}
-  
+
 	return dashborad, nil
 }
 
@@ -236,7 +233,7 @@ func (p *FunctionDefinitionsDeserializer) Unmarshal(data []byte) (interface{}, e
 		return nil, err
 	}
 	return functionDefinitions, nil
-  
+
 }
 
 type ShortUserInfosDeserializer struct{}
@@ -269,3 +266,35 @@ func (p *UserAccountDeserializer) Unmarshal(data []byte) (interface{}, error) {
 	return userAccount, nil
 
 }
+
+type PipelineDeserializer struct{}
+
+func NewPipelineDeserializer() *PipelineDeserializer {
+	return &PipelineDeserializer{}
+}
+
+func (p *PipelineDeserializer) Unmarshal(data []byte) (interface{}, error) {
+	var pipeLine pipeline.Pipeline
+	err := json.Unmarshal(data, &pipeLine)
+	if err != nil {
+		return nil, err
+	}
+	return pipeLine, nil
+
+}
+
+type PipelinesDeserializer struct{}
+
+func NewPipelinesDeserializer() *PipelinesDeserializer {
+	return &PipelinesDeserializer{}
+}
+
+func (p *PipelinesDeserializer) Unmarshal(data []byte) (interface{}, error) {
+	var pipeLine []pipeline.Pipeline
+	err := json.Unmarshal(data, &pipeLine)
+	if err != nil {
+		return nil, err
+	}
+	return pipeLine, nil
+
+}
diff --git a/streampipes-client-go/streampipes/internal/serializer/serializer.go b/streampipes-client-go/streampipes/internal/serializer/serializer.go
@@ -0,0 +1,37 @@
+//
+// 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 serializer
+
+import (
+	"encoding/json"
+	"github.com/apache/streampipes/streampipes-client-go/streampipes/model/pipeline"
+)
+
+type PipelineSerializer struct{}
+
+func NewPipelineSerializer() PipelineSerializer {
+	return PipelineSerializer{}
+}
+
+func (p PipelineSerializer) Marshal(pp pipeline.Pipeline) ([]byte, error) {
+	data, err := json.Marshal(pp)
+	if err != nil {
+		return nil, err
+	}
+	return data, nil
+}
diff --git a/streampipes-client-go/streampipes/model/common.go b/streampipes-client-go/streampipes/model/common.go
@@ -25,20 +25,22 @@ type ValueSpecification struct {
 }
 
 type EventProperty struct {
-	ElementID          string             `json:"elementId"`
-	Label              string             `json:"label,omitempty"`
-	Description        string             `json:"description,omitempty"`
-	RuntimeName        string             `json:"runtimeName,omitempty"`
-	Required           bool               `json:"required,omitempty"`
-	DomainProperties   []string           `json:"domainProperties,omitempty"`
-	PropertyScope      string             `json:"propertyScope,omitempty"`
-	Index              int                `json:"index"`
-	RuntimeID          string             `json:"runtimeId,omitempty"`
-	RuntimeType        string             `json:"runtimeType"`
-	MeasurementUnit    string             `json:"measurementUnit,omitempty"`
-	ValueSpecification ValueSpecification `json:"valueSpecification,omitempty"`
+	ElementID          string   `json:"elementId"`
+	Label              string   `json:"label,omitempty"`
+	Description        string   `json:"description,omitempty"`
+	RuntimeName        string   `json:"runtimeName,omitempty"`
+	Required           bool     `json:"required,omitempty"`
+	DomainProperties   []string `json:"domainProperties,omitempty"`
+	PropertyScope      string   `json:"propertyScope,omitempty"`
+	Index              int32    `json:"index"`
+	RuntimeID          string   `json:"runtimeId,omitempty"`
+	AdditionalMetadata map[string]interface{}
+	//RuntimeType        string             `json:"runtimeType"`
+	//MeasurementUnit    string             `json:"measurementUnit,omitempty"`
+	//ValueSpecification ValueSpecification `json:"valueSpecification,omitempty"`
 }
 
+// 有点问题，后续看一下
 type EventProperties struct {
 	ElementID          string            `json:"elementId"`
 	Label              string            `json:"label"`
@@ -47,7 +49,7 @@ type EventProperties struct {
 	Required           bool              `json:"required"`
 	DomainProperties   []string          `json:"domainProperties"`
 	PropertyScope      string            `json:"propertyScope"`
-	Index              int               `json:"index"`
+	Index              int32             `json:"index"`
 	RuntimeID          string            `json:"runtimeId"`
 	RuntimeType        string            `json:"runtimeType,omitempty"`
 	MeasurementUnit    string            `json:"measurementUnit,omitempty"`