You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampipes.apache.org by zi...@apache.org on 2024/03/23 15:32:53 UTC
(streampipes) branch dev updated: feat: add streampipes go client (#2437)
This is an automated email from the ASF dual-hosted git repository.
zike 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 7d3f3163d2 feat: add streampipes go client (#2437)
7d3f3163d2 is described below
commit 7d3f3163d232a7e11c18d1d7cce3101536740ac7
Author: wyyolo <10...@users.noreply.github.com>
AuthorDate: Sat Mar 23 23:32:49 2024 +0800
feat: add streampipes go client (#2437)
---
.github/workflows/go-client.yml | 35 ++++
streampipes-client-go/examples/main.go | 89 ++++++++++
streampipes-client-go/go.mod | 20 +++
.../config/streampipes_client_config.go | 42 +++++
.../streampipes/data_lake_measure_api.go | 184 +++++++++++++++++++++
streampipes-client-go/streampipes/endpoint.go | 73 ++++++++
.../streampipes/internal/http_headers/header.go | 54 ++++++
.../internal/serializer/deserializer.go | 76 +++++++++
.../internal/util/streampipes_api_path.go | 39 +++++
.../internal/util/streampipes_api_path_test.go | 52 ++++++
streampipes-client-go/streampipes/model/common.go | 68 ++++++++
.../model/data_lake/data_lake_measure.go | 32 ++++
.../streampipes/model/data_lake/data_series.go | 52 ++++++
.../streampipes/streampipes_client.go | 68 ++++++++
.../streampipes/utils/check_url.go | 25 +++
.../streampipes/utils/check_url_test.go | 52 ++++++
16 files changed, 961 insertions(+)
diff --git a/.github/workflows/go-client.yml b/.github/workflows/go-client.yml
new file mode 100644
index 0000000000..d967c2e830
--- /dev/null
+++ b/.github/workflows/go-client.yml
@@ -0,0 +1,35 @@
+# 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.
+
+name : go-client
+on:
+ pull_request:
+jobs:
+ go-unit-tests:
+ runs-on: ${{ matrix.os }}
+ strategy:
+ fail-fast: false
+ matrix:
+ go-version: [ '1.21.x' ]
+ os: [ubuntu-latest, macos-latest, windows-latest]
+ steps:
+ - uses: actions/checkout@v4
+ - uses: actions/setup-go@v4
+ with:
+ go-version: ${{ matrix.go-version }}
+ - name: Run unit-tests
+ working-directory: ./streampipes-client-go
+ run: |
+ go test ./...
diff --git a/streampipes-client-go/examples/main.go b/streampipes-client-go/examples/main.go
new file mode 100644
index 0000000000..cc053b8aff
--- /dev/null
+++ b/streampipes-client-go/examples/main.go
@@ -0,0 +1,89 @@
+//
+// 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 main
+
+import (
+ "github.com/apache/streampipes/streampipes-client-go/streampipes"
+ "github.com/apache/streampipes/streampipes-client-go/streampipes/config"
+ "log"
+)
+
+/*
+ Here are some examples of using go client, including outputting the data returned by streams.
+ Only supports outputting model data
+*/
+
+func main() {
+
+ /*
+ StreamPipesClientConfig can be configured in a number of ways.
+ For example：
+
+ First: clientConfig := config.StreamPipesClientConfig{
+ Url: "http://localhost:8030",
+ Credential: config.StreamPipesApiKeyCredentials{
+ UserName: "<Your-User-Name>",
+ ApiKey: "<Your-API-Key>",
+ },
+ }
+
+ The second : clientConfig := config.NewStreamPipesClientConnectConfig("http://localhost:8030",config.NewStreamPipesApiKeyCredentials("<Your-User-Name>","<Your-API-Key>"))
+
+ The third : clientConfig := := config.StreamPipesClientConfig{
+ Url: "http://localhost:8030",
+ Credential: config.NewStreamPipesApiKeyCredentials("<Your-User-Name>","<Your-API-Key>"),
+ }
+ */
+
+ clientConfig := config.StreamPipesClientConfig{
+ Url: "http://localhost:8030",
+ Credential: config.StreamPipesApiKeyCredentials{
+ UserName: "<Your-User-Name>",
+ ApiKey: "<Your-API-Key>",
+ },
+ }
+
+ streamPipesClient, err := streampipes.NewStreamPipesClient(clientConfig)
+ if err != nil {
+ log.Fatal(err)
+ }
+
+ dataSeries, err := streamPipesClient.DataLakeMeasures().GetSingleDataSeries("measureName")
+ if err != nil {
+ log.Fatal(err)
+ }
+ dataSeries.Print()
+
+ /*
+ output format:
+
+ There are 2 pieces of DataSerie in the Dataseries
+ The 1 DataSeries
+ time msg test
+ 2024-02-23T13:37:09.052Z go-client_test 2f4556
+ 2024-02-23T13:37:26.044Z go-client_test 2f4556
+ 2024-02-23T13:37:29.007Z go-client_test 2f4556
+ The 2 DataSeries
+ time msg test
+ 2024-02-23T13:38:06.052Z go-client_test 2f4556
+ 2024-02-23T13:38:35.044Z go-client_test 2f4556
+ 2024-02-23T13:38:38.007Z go-client_test 2f4556
+
+ */
+
+}
diff --git a/streampipes-client-go/go.mod b/streampipes-client-go/go.mod
new file mode 100644
index 0000000000..4cb7e16181
--- /dev/null
+++ b/streampipes-client-go/go.mod
@@ -0,0 +1,20 @@
+//
+// 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.
+//
+
+module github.com/apache/streampipes/streampipes-client-go
+
+go 1.21
diff --git a/streampipes-client-go/streampipes/config/streampipes_client_config.go b/streampipes-client-go/streampipes/config/streampipes_client_config.go
new file mode 100644
index 0000000000..b2006dd20e
--- /dev/null
+++ b/streampipes-client-go/streampipes/config/streampipes_client_config.go
@@ -0,0 +1,42 @@
+//
+// 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 config
+
+type StreamPipesClientConfig struct {
+ Url string
+ Credential StreamPipesApiKeyCredentials
+}
+
+type StreamPipesApiKeyCredentials struct {
+ UserName string
+ ApiKey string
+}
+
+func NewStreamPipesClientConnectConfig(url string, credentials StreamPipesApiKeyCredentials) StreamPipesClientConfig {
+ return StreamPipesClientConfig{
+ Url: url,
+ Credential: credentials,
+ }
+}
+
+func NewStreamPipesApiKeyCredentials(userName string, apiKey string) StreamPipesApiKeyCredentials {
+ return StreamPipesApiKeyCredentials{
+ UserName: userName,
+ ApiKey: apiKey,
+ }
+}
diff --git a/streampipes-client-go/streampipes/data_lake_measure_api.go b/streampipes-client-go/streampipes/data_lake_measure_api.go
new file mode 100644
index 0000000000..1a2fa9a5fb
--- /dev/null
+++ b/streampipes-client-go/streampipes/data_lake_measure_api.go
@@ -0,0 +1,184 @@
+//
+// 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/data_lake"
+ "io"
+ "log"
+ "net/http"
+)
+
+// DataLakeMeasure connects to the DataLakeMeasure endpoint of streamPipes.
+// DataLakeMeasure supports GET and DELETE to delete or obtain resources
+// The specific interaction behavior is provided by the method bound to the DataLakeMeasure struct.
+type DataLakeMeasure struct {
+ endpoint
+}
+
+func NewDataLakeMeasures(clientConfig config.StreamPipesClientConfig) *DataLakeMeasure {
+ // NewDataLakeMeasure is used to return an instance of *DataLakeMeasure,
+
+ return &DataLakeMeasure{
+ endpoint{config: clientConfig},
+ }
+}
+
+// AllDataLakeMeasure retrieves all the measures from the Data Lake.
+func (d *DataLakeMeasure) AllDataLakeMeasure() ([]data_lake.DataLakeMeasure, error) {
+
+ endPointUrl := util.NewStreamPipesApiPath(d.config.Url, "streampipes-backend/api/v4/datalake/measurements", nil)
+ log.Printf("Get data from: %s", endPointUrl)
+
+ response, err := d.executeRequest("GET", endPointUrl)
+ if err != nil {
+ return nil, err
+ }
+
+ if response.StatusCode != http.StatusOK {
+ err = d.handleStatusCode(response)
+ if err != nil {
+ return nil, err
+ }
+ }
+
+ body, err := io.ReadAll(response.Body)
+ if err != nil {
+ return nil, err
+ }
+
+ unmarshalData, err := serializer.NewDataLakeMeasuresDeserializer().Unmarshal(body)
+ if err != nil {
+ return nil, err
+ }
+ dataLakeMeasures := unmarshalData.([]data_lake.DataLakeMeasure)
+
+ return dataLakeMeasures, nil
+}
+
+// GetSingleDataLakeMeasure retrieves a specific measure from the Data Lake.
+func (d *DataLakeMeasure) GetSingleDataLakeMeasure(elementId string) (data_lake.DataLakeMeasure, error) {
+
+ endPointUrl := util.NewStreamPipesApiPath(d.config.Url, "streampipes-backend/api/v4/datalake/measure", []string{elementId})
+ log.Printf("Get data from: %s", endPointUrl)
+
+ response, err := d.executeRequest("GET", endPointUrl)
+ if err != nil {
+ return data_lake.DataLakeMeasure{}, err
+ }
+
+ if response.StatusCode != http.StatusOK {
+ err = d.handleStatusCode(response)
+ if err != nil {
+ return data_lake.DataLakeMeasure{}, err
+ }
+ }
+
+ body, err := io.ReadAll(response.Body)
+ if err != nil {
+ return data_lake.DataLakeMeasure{}, err
+ }
+
+ unmarshalData, err := serializer.NewDataLakeMeasureDeserializer().Unmarshal(body)
+ if err != nil {
+ return data_lake.DataLakeMeasure{}, err
+ }
+ dataLakeMeasure := unmarshalData.(data_lake.DataLakeMeasure)
+
+ return dataLakeMeasure, nil
+}
+
+// GetSingleDataSeries retrieves the measurement series for the specified measureId from the Data Lake.
+// Currently not supporting parameter queries.
+// The measureId can also be considered measureName.
+func (d *DataLakeMeasure) GetSingleDataSeries(measureId string) (*data_lake.DataSeries, error) {
+
+ endPointUrl := util.NewStreamPipesApiPath(d.config.Url, "streampipes-backend/api/v4/datalake/measurements", []string{measureId})
+ log.Printf("Get data from: %s", endPointUrl)
+
+ response, err := d.executeRequest("GET", endPointUrl)
+ if err != nil {
+ return nil, err
+ }
+
+ if response.StatusCode != http.StatusOK {
+ err = d.handleStatusCode(response)
+ if err != nil {
+ return nil, err
+ }
+ }
+
+ body, err := io.ReadAll(response.Body)
+ if err != nil {
+ return nil, err
+ }
+
+ unmarshalData, err := serializer.NewDataSeriesDeserializer().Unmarshal(body)
+ if err != nil {
+ return nil, err
+ }
+ dataSeries := unmarshalData.(data_lake.DataSeries)
+
+ return &dataSeries, nil
+}
+
+// ClearDataLakeMeasureData removes data from a single measurement series with given id.
+func (d *DataLakeMeasure) ClearDataLakeMeasureData(measureId string) error {
+
+ endPointUrl := util.NewStreamPipesApiPath(d.config.Url, "streampipes-backend/api/v4/datalake/measurements", []string{measureId})
+ log.Printf("Clear data from: %s", endPointUrl)
+
+ response, err := d.executeRequest("DELETE", endPointUrl)
+ if err != nil {
+ return err
+ }
+
+ if response.StatusCode != http.StatusOK {
+ err = d.handleStatusCode(response)
+ if err != nil {
+ return err
+ }
+ }
+ log.Printf("Successfully deleted data from a single measurement sequence of %s", measureId)
+
+ return nil
+}
+
+// DeleteDataLakeMeasure drops a single measurement series with given id from Data Lake and remove related event property.
+func (d *DataLakeMeasure) DeleteDataLakeMeasure(measureId string) error {
+
+ endPointUrl := util.NewStreamPipesApiPath(d.config.Url, "streampipes-backend/api/v4/datalake/measurements", []string{measureId, "drop"})
+ log.Printf("Delete data from: %s", endPointUrl)
+ response, err := d.executeRequest("DELETE", endPointUrl)
+ if err != nil {
+ return err
+ }
+
+ if response.StatusCode != http.StatusOK {
+ err = d.handleStatusCode(response)
+ if err != nil {
+ return err
+ }
+ }
+
+ log.Printf("Successfully dropped a single measurement series for %s from DataLake and remove related event property", measureId)
+ return nil
+}
diff --git a/streampipes-client-go/streampipes/endpoint.go b/streampipes-client-go/streampipes/endpoint.go
new file mode 100644
index 0000000000..2e3790dc93
--- /dev/null
+++ b/streampipes-client-go/streampipes/endpoint.go
@@ -0,0 +1,73 @@
+//
+// 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 (
+ "errors"
+ "github.com/apache/streampipes/streampipes-client-go/streampipes/config"
+ headers "github.com/apache/streampipes/streampipes-client-go/streampipes/internal/http_headers"
+ "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)
+ if err != nil {
+ return nil, err
+ }
+
+ header := headers.NewHeaders(req)
+ header.SetApiUserAndApiKey(e.config.Credential.UserName, e.config.Credential.ApiKey)
+ header.SetAcceptJson()
+ header.SetContentTypeJson()
+
+ client := &http.Client{}
+ response, err := client.Do(header.GetReq())
+ if err != nil {
+ return nil, err
+ }
+ return response, nil
+}
+
+func (e *endpoint) handleStatusCode(resp *http.Response) error {
+
+ switch resp.StatusCode {
+ case http.StatusUnauthorized:
+ return errors.New("The streamPipes Backend returned an unauthorized error.\nplease check your ApiUser and/or Apikey to be correct.")
+ case http.StatusForbidden:
+ return errors.New("There seems to be an issue with the access rights of the given user and the resource you queried.\n" +
+ "Apparently, this user is not allowed to query the resource.\n" +
+ "Please check the user's permissions or contact your StreamPipes admin.")
+ case http.StatusNotFound:
+ return errors.New("There seems to be an issue with the Go Client calling the API inappropriately.\n" +
+ "This should not happen, but unfortunately did.\n" +
+ "If you don't mind, it would be awesome to let us know by creating an issue at https://github.com/apache/streampipes.\n")
+ case http.StatusMethodNotAllowed:
+ return errors.New("There seems to be an issue with the Go Client calling the API inappropriately.\n" +
+ "This should not happen, but unfortunately did.\n" +
+ "If you don't mind, it would be awesome to let us know by creating an issue at https://github.com/apache/streampipes.\n")
+ case http.StatusInternalServerError:
+ return errors.New("streamPipes internal error")
+ default:
+ return errors.New(resp.Status)
+ }
+
+}
diff --git a/streampipes-client-go/streampipes/internal/http_headers/header.go b/streampipes-client-go/streampipes/internal/http_headers/header.go
new file mode 100644
index 0000000000..c201078c80
--- /dev/null
+++ b/streampipes-client-go/streampipes/internal/http_headers/header.go
@@ -0,0 +1,54 @@
+//
+// 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 http_headers
+
+import (
+ "net/http"
+)
+
+type Headers struct {
+ req *http.Request
+ header http.Header
+}
+
+func NewHeaders(req *http.Request) *Headers {
+ return &Headers{
+ req: req,
+ header: nil,
+ }
+}
+
+func (h *Headers) SetApiUserAndApiKey(apiUser string, apiKey string) {
+ h.header = h.req.Header
+ h.header.Set("X-API-USER", apiUser)
+ h.header.Set("X-API-KEY", apiKey)
+}
+
+func (h *Headers) SetAcceptJson() {
+ h.header = h.req.Header
+ h.header.Set("Accept", "application/json")
+}
+
+func (h *Headers) SetContentTypeJson() {
+ h.header = h.req.Header
+ h.header.Set("Content-type", "application/json")
+}
+
+func (h *Headers) GetReq() *http.Request {
+ return h.req
+}
diff --git a/streampipes-client-go/streampipes/internal/serializer/deserializer.go b/streampipes-client-go/streampipes/internal/serializer/deserializer.go
new file mode 100644
index 0000000000..94946f6564
--- /dev/null
+++ b/streampipes-client-go/streampipes/internal/serializer/deserializer.go
@@ -0,0 +1,76 @@
+//
+// 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/data_lake"
+)
+
+type Deserializer interface {
+ Unmarshal(body []byte) (interface{}, error)
+}
+
+var _ Deserializer = (*DataLakeMeasuresDeserializer)(nil)
+var _ Deserializer = (*DataSeriesDeserializer)(nil)
+var _ Deserializer = (*DataLakeMeasureDeserializer)(nil)
+
+type DataLakeMeasuresDeserializer struct{}
+
+func NewDataLakeMeasuresDeserializer() *DataLakeMeasuresDeserializer {
+ return &DataLakeMeasuresDeserializer{}
+}
+
+func (u *DataLakeMeasuresDeserializer) Unmarshal(data []byte) (interface{}, error) {
+ var dataLakeMeasures []data_lake.DataLakeMeasure
+ err := json.Unmarshal(data, &dataLakeMeasures)
+ if err != nil {
+ return nil, err
+ }
+ return dataLakeMeasures, nil
+}
+
+type DataLakeMeasureDeserializer struct{}
+
+func NewDataLakeMeasureDeserializer() *DataLakeMeasureDeserializer {
+ return &DataLakeMeasureDeserializer{}
+}
+
+func (u *DataLakeMeasureDeserializer) Unmarshal(data []byte) (interface{}, error) {
+ var dataLakeMeasure data_lake.DataLakeMeasure
+ err := json.Unmarshal(data, &dataLakeMeasure)
+ if err != nil {
+ return nil, err
+ }
+ return dataLakeMeasure, nil
+}
+
+type DataSeriesDeserializer struct{}
+
+func NewDataSeriesDeserializer() *DataSeriesDeserializer {
+ return &DataSeriesDeserializer{}
+}
+
+func (u *DataSeriesDeserializer) Unmarshal(data []byte) (interface{}, error) {
+ var dataSeries data_lake.DataSeries
+ err := json.Unmarshal(data, &dataSeries)
+ if err != nil {
+ return nil, err
+ }
+ return dataSeries, nil
+}
diff --git a/streampipes-client-go/streampipes/internal/util/streampipes_api_path.go b/streampipes-client-go/streampipes/internal/util/streampipes_api_path.go
new file mode 100644
index 0000000000..6a6fd7fcac
--- /dev/null
+++ b/streampipes-client-go/streampipes/internal/util/streampipes_api_path.go
@@ -0,0 +1,39 @@
+//
+// 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 util
+
+import (
+ "strings"
+)
+
+func NewStreamPipesApiPath(baseUrl string, streamPipesUrl string, parameters []string) string {
+ path := []string{baseUrl, streamPipesUrl}
+ path = append(path, parameters...)
+ return join(path)
+}
+
+func join(path []string) string {
+ if path == nil {
+ return ""
+ }
+ if len(path) == 1 {
+ return path[0]
+ }
+ url := strings.Join(path, "/")
+ return url
+}
diff --git a/streampipes-client-go/streampipes/internal/util/streampipes_api_path_test.go b/streampipes-client-go/streampipes/internal/util/streampipes_api_path_test.go
new file mode 100644
index 0000000000..c17f9de4ec
--- /dev/null
+++ b/streampipes-client-go/streampipes/internal/util/streampipes_api_path_test.go
@@ -0,0 +1,52 @@
+//
+// 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 util
+
+import "testing"
+
+func TestJoin(t *testing.T) {
+ testCases := []struct {
+ name string
+ input []string
+ expected string
+ }{
+ {
+ name: "Single URL",
+ input: []string{"http://localhost:8030"},
+ expected: "http://localhost:8030",
+ },
+ {
+ name: "Empty slice",
+ input: []string{},
+ expected: "",
+ },
+ {
+ name: "Multiple URL parts",
+ input: []string{"http://localhost:8030", "test1", "test2"},
+ expected: "http://localhost:8030/test1/test2",
+ },
+ }
+ for _, tc := range testCases {
+ t.Run(tc.name, func(t *testing.T) {
+ result := join(tc.input)
+ if result != tc.expected {
+ t.Errorf("Expected '%s', got '%s'", tc.expected, result)
+ }
+ })
+ }
+}
diff --git a/streampipes-client-go/streampipes/model/common.go b/streampipes-client-go/streampipes/model/common.go
new file mode 100644
index 0000000000..f0ba11637a
--- /dev/null
+++ b/streampipes-client-go/streampipes/model/common.go
@@ -0,0 +1,68 @@
+//
+// 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 model
+
+type ValueSpecification struct {
+ ElementID string `json:"elementId,omitempty"`
+ MinValue int `json:"minValue,omitempty"`
+ MaxValue int `json:"maxValue,omitempty"`
+ Step float64 `json:"step,omitempty"`
+}
+
+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"`
+}
+
+type EventProperties struct {
+ ElementID string `json:"elementId"`
+ Label string `json:"label"`
+ Description string `json:"description"`
+ RuntimeName string `json:"runtimeName"`
+ Required bool `json:"required"`
+ DomainProperties []string `json:"domainProperties"`
+ PropertyScope string `json:"propertyScope"`
+ Index int `json:"index"`
+ RuntimeID string `json:"runtimeId"`
+ RuntimeType string `json:"runtimeType,omitempty"`
+ MeasurementUnit string `json:"measurementUnit,omitempty"`
+ ValueSpecification string `json:"valueSpecification,omitempty"`
+ InEventProperties []EventProperties `json:"eventProperties,omitempty"`
+ InEventProperty EventProperty `json:"eventProperty,omitempty"`
+}
+
+type EventSchema struct {
+ EventProperties []EventProperties `json:"eventProperties"`
+}
+
+type DataSeries struct {
+ Total int `json:"total"`
+ Rows [][]string `json:"rows"`
+ Headers []string `json:"http_headers"`
+ Tags map[string]string `json:"tags"`
+}
diff --git a/streampipes-client-go/streampipes/model/data_lake/data_lake_measure.go b/streampipes-client-go/streampipes/model/data_lake/data_lake_measure.go
new file mode 100644
index 0000000000..2e1c48da45
--- /dev/null
+++ b/streampipes-client-go/streampipes/model/data_lake/data_lake_measure.go
@@ -0,0 +1,32 @@
+//
+// 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 data_lake
+
+import "github.com/apache/streampipes/streampipes-client-go/streampipes/model"
+
+type DataLakeMeasure struct {
+ MeasureName string `json:"measureName"`
+ TimestampField string `json:"timestampField"`
+ EventSchema model.EventSchema `json:"eventSchema,omitempty"`
+ PipelineId string `json:"pipelineId,omitempty"`
+ PipelineName string `json:"pipelineName,omitempty"`
+ PipelineIsRunning bool `json:"pipelineIsRunning"`
+ SchemaVersion string `json:"schemaVersion,omitempty"`
+ SchemaUpdateStrategy string `json:"schemaUpdateStrategy,omitempty"`
+ ElementId string `json:"elementId"`
+}
diff --git a/streampipes-client-go/streampipes/model/data_lake/data_series.go b/streampipes-client-go/streampipes/model/data_lake/data_series.go
new file mode 100644
index 0000000000..b273b42280
--- /dev/null
+++ b/streampipes-client-go/streampipes/model/data_lake/data_series.go
@@ -0,0 +1,52 @@
+//
+// 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 data_lake
+
+import (
+ "fmt"
+ "github.com/apache/streampipes/streampipes-client-go/streampipes/model"
+ "log"
+ "strings"
+)
+
+type DataSeries struct {
+ Total int `json:"total"`
+ Headers []string `json:"http_headers"`
+ AllDataSeries []model.DataSeries `json:"allDataSeries"`
+ SPQueryStatus string `alias:"spQueryStatus" default:"OK"`
+ ForId interface{} `json:"forId"`
+}
+
+func (d *DataSeries) Print() {
+ if d.Total == 0 {
+ log.Println("This DataSeries has no data")
+ return
+ }
+ fmt.Printf("There are %d pieces of DataSerie in the Dataseries\n", d.Total)
+ for i := 0; i < d.Total; i++ {
+ fmt.Printf("The %d DataSeries\n", i+1)
+ headers := strings.Join(d.Headers, " ")
+ fmt.Printf("%s\n", headers)
+ for j := 0; j < d.AllDataSeries[i].Total; j++ {
+ singleLine := strings.Join(d.AllDataSeries[i].Rows[j], " ")
+ fmt.Printf("%s\n",
+ singleLine,
+ )
+ }
+ }
+}
diff --git a/streampipes-client-go/streampipes/streampipes_client.go b/streampipes-client-go/streampipes/streampipes_client.go
new file mode 100644
index 0000000000..3c6ea098e1
--- /dev/null
+++ b/streampipes-client-go/streampipes/streampipes_client.go
@@ -0,0 +1,68 @@
+//
+// 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 (
+ "errors"
+ "github.com/apache/streampipes/streampipes-client-go/streampipes/config"
+ "github.com/apache/streampipes/streampipes-client-go/streampipes/utils"
+ "net/url"
+ "strings"
+)
+
+// This is the central point of contact with StreamPipes and provides all the functionalities to interact with it.
+// The client provides so-called "API", each of which refers to the endpoint of the StreamPipes API.
+// e.g. `DataLakeMeasure` provides the actual methods to interact with StreamPipes API.
+
+type StreamPipesClient struct {
+ config config.StreamPipesClientConfig
+}
+
+// NewStreamPipesClient returns an instance of *StreamPipesClient
+// Currently, it does not support HTTPS connections or connections to port 443.
+func NewStreamPipesClient(c config.StreamPipesClientConfig) (*StreamPipesClient, error) {
+
+ if c.Credential == (config.StreamPipesApiKeyCredentials{}) {
+ return nil, errors.New("no credential entered")
+ }
+
+ if !utils.CheckUrl(c.Url) {
+ return nil, errors.New("please check if the URL is correct,Must be in the form of A://B:C," +
+ "where A is either HTTP, not case sensitive. B must be the host and C must be the port")
+ }
+
+ Url, err := url.Parse(c.Url)
+ if err != nil {
+ return nil, err
+ }
+
+ if strings.EqualFold(Url.Scheme, "https") || Url.Port() == "443" {
+ return nil, errors.New(
+ "The URL passed in is invalid and does not support HTTPS or port 443.\n ")
+ }
+
+ return &StreamPipesClient{
+ c,
+ }, nil
+
+}
+
+func (s *StreamPipesClient) DataLakeMeasures() *DataLakeMeasure {
+
+ return NewDataLakeMeasures(s.config)
+}
diff --git a/streampipes-client-go/streampipes/utils/check_url.go b/streampipes-client-go/streampipes/utils/check_url.go
new file mode 100644
index 0000000000..d3a47431a3
--- /dev/null
+++ b/streampipes-client-go/streampipes/utils/check_url.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 utils
+
+import "regexp"
+
+func CheckUrl(url string) bool {
+ re := regexp.MustCompile(`(?i)^(http|https):\/\/[^\s\/:]+:\d+$`)
+ return re.MatchString(url)
+}
diff --git a/streampipes-client-go/streampipes/utils/check_url_test.go b/streampipes-client-go/streampipes/utils/check_url_test.go
new file mode 100644
index 0000000000..adb1f551d5
--- /dev/null
+++ b/streampipes-client-go/streampipes/utils/check_url_test.go
@@ -0,0 +1,52 @@
+//
+// 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 utils
+
+import (
+ "testing"
+)
+
+func TestCheckUrl(t *testing.T) {
+ tests := []struct {
+ name string
+ url string
+ expected bool
+ }{
+ {name: "Valid HTTP URL", url: "http://streampipes.com:80", expected: true},
+ {name: "Valid HTTP URL", url: "http://localhost:80", expected: true},
+ {name: "Valid HTTP URL", url: "http://127.0.0.1:80", expected: true},
+ {name: "Valid HTTPS URL", url: "https://streampipes.com:443", expected: true},
+ {name: "Valid URL with port", url: "http://streampipes.com:8080", expected: true},
+ {name: "Valid URL with IPv6", url: "http://[2001:db8:85a3:8d3:1319:8a2e:370:7348]:80", expected: false},
+ {name: "Invalid URL scheme", url: "ftp://streampipes.com:80", expected: false},
+ {name: "Invalid URL with no port", url: "http://streampipes.com", expected: false},
+ {name: "Invalid URL with space", url: "http://streampipes.com: 80", expected: false},
+ {name: "Invalid URL with special character", url: "http://streampipes.com:$80", expected: false},
+ {name: "Empty URL", url: "", expected: false},
+ {name: "Nil URL", url: "", expected: false},
+ }
+
+ for _, test := range tests {
+ t.Run(test.name, func(t *testing.T) {
+ actual := CheckUrl(test.url)
+ if actual != test.expected {
+ t.Errorf("CheckUrl(%s) = %v; want %v", test.url, actual, test.expected)
+ }
+ })
+ }
+}