style: align with gofumpt format style

This is a preprocessed PR for introducing golangci-lint.

Signed-off-by: xjasonlyu <xjasonlyu@gmail.com>

internal/binder/mock/mock_factory.go

@@ -1,4 +1,4 @@
-// Copyright 2021-2022 EMQ Technologies Co., Ltd.
+// Copyright 2021-2023 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -15,14 +15,14 @@
 package mock
 
 import (
+	"strings"
+
 	"github.com/lf-edge/ekuiper/internal/plugin"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/errorx"
-	"strings"
 )
 
-type MockFactory struct {
-}
+type MockFactory struct{}
 
 func NewMockFactory() *MockFactory {
 	return &MockFactory{}

internal/converter/binary/converter.go

@@ -1,4 +1,4 @@
-// Copyright 2022 EMQ Technologies Co., Ltd.
+// Copyright 2022-2023 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -16,11 +16,11 @@ package binary
 
 import (
 	"fmt"
+
 	"github.com/lf-edge/ekuiper/pkg/message"
 )
 
-type Converter struct {
-}
+type Converter struct{}
 
 var converter = &Converter{}
 

internal/converter/binary/converter_test.go

@@ -26,7 +26,7 @@ func TestMessageDecode(t *testing.T) {
 	if err != nil {
 		t.Errorf("Cannot read image: %v", err)
 	}
-	var tests = []struct {
+	tests := []struct {
 		payload []byte
 		result  map[string]interface{}
 	}{

internal/converter/json/converter.go

@@ -1,4 +1,4 @@
-// Copyright 2022 EMQ Technologies Co., Ltd.
+// Copyright 2022-2023 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -16,11 +16,11 @@ package json
 
 import (
 	"encoding/json"
+
 	"github.com/lf-edge/ekuiper/pkg/message"
 )
 
-type Converter struct {
-}
+type Converter struct{}
 
 var converter = &Converter{}
 

internal/converter/json/converter_test.go

@@ -29,7 +29,7 @@ func TestMessageDecode(t *testing.T) {
 		t.Errorf("Cannot read image: %v", err)
 	}
 	b64img := base64.StdEncoding.EncodeToString(image)
-	var tests = []struct {
+	tests := []struct {
 		payload []byte
 		format  string
 		result  map[string]interface{}

internal/io/sink/log_sink.go

@@ -16,10 +16,11 @@ package sink
 
 import (
 	"fmt"
-	"github.com/lf-edge/ekuiper/internal/topo/collector"
-	"github.com/lf-edge/ekuiper/pkg/api"
 	"sync"
 	"time"
+
+	"github.com/lf-edge/ekuiper/internal/topo/collector"
+	"github.com/lf-edge/ekuiper/pkg/api"
 )
 
 // NewLogSink log action, no properties now

internal/io/sink/nop_sink.go

@@ -23,7 +23,7 @@ type NopSink struct {
 }
 
 func (ns *NopSink) Configure(ps map[string]interface{}) error {
-	var log = false
+	var log bool
 	l, ok := ps["log"]
 	if ok {
 		log = l.(bool)

internal/pkg/filex/parser.go

@@ -1,4 +1,4 @@
-// Copyright 2021 EMQ Technologies Co., Ltd.
+// Copyright 2021-2023 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -32,6 +32,7 @@ func ReadJsonUnmarshal(path string, ret interface{}) error {
 	}
 	return nil
 }
+
 func WriteYamlMarshal(path string, data interface{}) error {
 	y, err := yaml.Marshal(data)
 	if nil != err {

internal/topo/collector/func.go

@@ -1,4 +1,4 @@
-// Copyright 2021 EMQ Technologies Co., Ltd.
+// Copyright 2021-2023 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -16,6 +16,7 @@ package collector
 
 import (
 	"errors"
+
 	"github.com/lf-edge/ekuiper/pkg/api"
 )
 
@@ -41,7 +42,7 @@ func Func(f CollectorFunc) *FuncCollector {
 }
 
 func (c *FuncCollector) Configure(props map[string]interface{}) error {
-	//do nothing
+	// do nothing
 	return nil
 }
 

internal/topo/graph/io.go

@@ -16,9 +16,11 @@ package graph
 
 import "fmt"
 
-type IoInputType uint8
-type IoRowType uint8
-type IoCollectionType uint8
+type (
+	IoInputType      uint8
+	IoRowType        uint8
+	IoCollectionType uint8
+)
 
 const (
 	IOINPUT_TYPE_SAME       IoInputType = iota

pkg/api/stream.go

@@ -39,6 +39,7 @@ func NewDefaultSourceTuple(message map[string]interface{}, meta map[string]inter
 		Time: time.Now(),
 	}
 }
+
 func NewDefaultSourceTupleWithTime(message map[string]interface{}, meta map[string]interface{}, timestamp time.Time) *DefaultSourceTuple {
 	return &DefaultSourceTuple{
 		Mess: message,
@@ -50,6 +51,7 @@ func NewDefaultSourceTupleWithTime(message map[string]interface{}, meta map[stri
 func (t *DefaultSourceTuple) Message() map[string]interface{} {
 	return t.Mess
 }
+
 func (t *DefaultSourceTuple) Meta() map[string]interface{} {
 	return t.M
 }
@@ -75,7 +77,8 @@ type Logger interface {
 
 type Store interface {
 	SaveState(checkpointId int64, opId string, state map[string]interface{}) error
-	SaveCheckpoint(checkpointId int64) error //Save the whole checkpoint state into storage
+	// SaveCheckpoint saves the whole checkpoint state into storage
+	SaveCheckpoint(checkpointId int64) error
 	GetOpState(opId string) (*sync.Map, error)
 	Clean() error
 }
@@ -88,7 +91,7 @@ type Source interface {
 	// Open Should be sync function for normal case. The container will run it in go func
 	Open(ctx StreamContext, consumer chan<- SourceTuple, errCh chan<- error)
 	// Configure Called during initialization. Configure the source with the data source(e.g. topic for mqtt) and the properties
-	//read from the yaml
+	// read from the yaml
 	Configure(datasource string, props map[string]interface{}) error
 	Closable
 }
@@ -97,7 +100,7 @@ type LookupSource interface {
 	// Open creates the connection to the external data source
 	Open(ctx StreamContext) error
 	// Configure Called during initialization. Configure the source with the data source(e.g. topic for mqtt) and the properties
-	//read from the yaml
+	// read from the yaml
 	Configure(datasource string, props map[string]interface{}) error
 	// Lookup receive lookup values to construct the query and return query results
 	Lookup(ctx StreamContext, fields []string, keys []string, values []interface{}) ([]SourceTuple, error)
@@ -160,7 +163,8 @@ type GraphNode struct {
 	Type     string                 `json:"type"`
 	NodeType string                 `json:"nodeType"`
 	Props    map[string]interface{} `json:"props"`
-	UI       map[string]interface{} `json:"ui"` //placeholder for ui properties
+	// UI is a placeholder for ui properties
+	UI map[string]interface{} `json:"ui"`
 }
 
 type RuleGraph struct {
@@ -231,7 +235,7 @@ type Function interface {
 	// Validate The argument is a list of xsql.Expr
 	Validate(args []interface{}) error
 	// Exec Execute the function, return the result and if execution is successful.
-	//If execution fails, return the error and false.
+	// If execution fails, return the error and false.
 	Exec(args []interface{}, ctx FunctionContext) (interface{}, bool)
 	// IsAggregate If this function is an aggregate function. Each parameter of an aggregate function will be a slice
 	IsAggregate() bool

pkg/kv/kv.go

@@ -1,4 +1,4 @@
-// Copyright 2021-2022 EMQ Technologies Co., Ltd.
+// Copyright 2021-2023 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -15,14 +15,14 @@
 package kv
 
 type KeyValue interface {
-	// Set key to hold string value if key does not exist otherwise return an error
+	// Setnx sets key to hold string value if key does not exist otherwise return an error
 	Setnx(key string, value interface{}) error
 	// Set key to hold the string value. If key already holds a value, it is overwritten
 	Set(key string, value interface{}) error
 	Get(key string, val interface{}) (bool, error)
 	GetKeyedState(key string) (interface{}, error)
 	SetKeyedState(key string, value interface{}) error
-	//Must return *common.Error with NOT_FOUND error
+	// Delete must return *common.Error with NOT_FOUND error
 	Delete(key string) error
 	Keys() (keys []string, err error)
 	All() (all map[string]string, err error)

test/plugins/pub/zmq_pub.go

@@ -1,4 +1,4 @@
-// Copyright 2021 EMQ Technologies Co., Ltd.
+// Copyright 2021-2023 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -17,9 +17,10 @@ package main
 import (
 	"encoding/json"
 	"fmt"
-	zmq "github.com/pebbe/zmq4"
 	"os"
 	"time"
+
+	zmq "github.com/pebbe/zmq4"
 )
 
 type zmqPub struct {