refacor: change config field downstreams to mesh (#694)

# Description

Change config field `downstreams` to `mesh`, The `mesh` contains all
information of zippers.

cli/serve.go

@@ -63,7 +63,7 @@ var serveCmd = &cobra.Command{
 			}
 		}
 
-		zipper, err := yomo.NewZipper(conf.Name, router.Default(), conf.Downstreams, options...)
+		zipper, err := yomo.NewZipper(conf.Name, router.Default(), conf.Mesh, options...)
 		if err != nil {
 			log.FailureStatusEvent(os.Stdout, err.Error())
 			return

docs/pages/docs/cli/zipper.mdx

@@ -16,7 +16,7 @@ the `config.yaml` is a YAML file, which contains the configuration of the Zipper
 
 ```yaml filename="config.yaml"
 ### general ###
-name: america
+name: zipper-sgp
 host: 0.0.0.0
 port: 9000
 
@@ -25,20 +25,24 @@ auth:
   type: token
   token: <CREDENTIAL>
 
-### cascading ###
-downstreams:
-   - zipper-asia_pacific
-     host: 2.2.2.2
-     port: 9000
-     credential: 'token: <CREDENTIAL>'
-   - zipper-america
-     host: 3.3.3.3
-     port: 9000
-     credential: 'token: <CREDENTIAL>'
-   - zipper-europe
-     host: 4.4.4.4
-     port: 9000
-     credential: 'token: <CREDENTIAL>'
+### cascading mesh ###
+mesh:
+  zipper-sgp:
+    host: 1.1.1.1
+    port: 9000
+    credential: "token: <CREDENTIAL>"
+  zipper-aus:
+    host: 2.2.2.2
+    port: 9000
+    credential: "token: <CREDENTIAL>"
+  zipper-usa:
+    host: 3.3.3.3
+    port: 9000
+    auth: "token: <CREDENTIAL>"
+  zipper-deu:
+    host: 4.4.4.4
+    port: 9000
+    auth: "token: <CREDENTIAL>"
 ```
 
 ## Configuration
@@ -55,13 +59,13 @@ downstreams:
   - `type` - the type of the credential, currently only `token` is supported.
   - `token` - the credential, it is a string.
 
-### Downstreams Config
+### Mesh Config
 
-- `downstreams` - the list of downstream Zippers, the data will be forwarded to these Zippers when needed.
-  - `name` - the name of the downstream Zipper, it is used to identify the Zipper in the network.
-  - `host` - the IP address of the downstream Zipper.
-  - `port` - the port of the downstream Zipper.
-  - `credential` - the credential to connect to the downstream Zipper.
+- `mesh` - the list of mesh Zippers, the data will be transmitted between these Zippers when needed.
+  - `name` - the name of the mesh Zipper, it is used to identify the Zipper in the network.
+  - `host` - the IP address of the mesh Zipper.
+  - `port` - the port of the mesh Zipper.
+  - `credential` - the credential to connect to the mesh Zipper.
 
 ## Self-Hosting Zipper Service
 

example/4-cascading-zipper/zipper_1.yaml

@@ -4,7 +4,12 @@ port: 9001
 auth:
   type: token
   token: z1
-downstreams:
+
+mesh:
+  zipper-1:
+    host: 127.0.0.1
+    port: 9001
+    credential: "token:z1"
   zipper-2:
     host: 127.0.0.1
     port: 9002

example/4-cascading-zipper/zipper_2.yaml

@@ -4,8 +4,13 @@ port: 9002
 auth:
   type: token
   token: z2
-downstreams:
+
+mesh:
   zipper-1:
     host: 127.0.0.1
     port: 9001
     credential: "token:z1"
+  zipper-2:
+    host: 127.0.0.1
+    port: 9002
+    credential: "token:z2"

example/6-mesh/eu/sender.yaml

@@ -1,7 +1,11 @@
 name: Zipper-Sender-EU
 host: localhost
 port: 9800
-downstreams:
+
+mesh:
+  Zipper-Sender-EU:
+    host: localhost
+    port: 9800
   Zipper-Receiver-US:
     host: localhost
     port: 9001

example/6-mesh/us/sender.yaml

@@ -1,7 +1,11 @@
 name: Zipper-Sender-US
 host: localhost
 port: 9000
-downstreams:
+
+mesh:
+  Zipper-Sender-US:
+    host: localhost
+    port: 9000
   Zipper-Receiver-US:
     host: localhost
     port: 9001

pkg/config/config.go

@@ -27,19 +27,19 @@ type Config struct {
 	// The token typed auth has two key-value pairs associated with it:
 	// a `type:token` key-value pair and a `token:<CREDENTIAL>` key-value pair.
 	Auth map[string]string `yaml:"auth"`
-	// Downstreams holds cascading zippers config. the map-key is downstream name.
-	Downstreams map[string]Downstream `yaml:"downstreams"`
+	// Mesh holds all cascading zippers config. the map-key is mesh name.
+	Mesh map[string]Mesh `yaml:"mesh"`
 }
 
-// Downstream describes a cascading zipper config.
-type Downstream struct {
-	// Host is the host of downstream zipper.
+// Mesh describes a cascading zipper config.
+type Mesh struct {
+	// Host is the host of mesh zipper.
 	Host string `yaml:"host"`
-	// Port is the port of downstream zipper.
+	// Port is the port of mesh zipper.
 	Port int `yaml:"port"`
-	// Credential is the credential when connect to downstream zipper.
+	// Credential is the credential when connect to mesh zipper.
 	// It is in the format of 'authType:authPayload', separated by a colon.
-	// If Credential is empty, it represents that downstream will not authenticate the current Zipper.
+	// If Credential is empty, it represents that mesh will not authenticate the current Zipper.
 	Credential string `yaml:"credential"`
 }
 

pkg/config/config_test.go

@@ -20,7 +20,7 @@ func TestParseConfigFile(t *testing.T) {
 		conf, err := ParseConfigFile("../../test/config.yaml")
 		assert.NoError(t, err)
 
-		assert.Equal(t, "zipper-chn", conf.Name)
+		assert.Equal(t, "zipper-sgp", conf.Name)
 		assert.Equal(t, "0.0.0.0", conf.Host)
 
 		assert.Equal(t, 9000, conf.Port)

test/config.yaml

@@ -1,5 +1,5 @@
 ### general ###
-name: zipper-chn
+name: zipper-sgp
 host: 0.0.0.0
 port: 9000
 
@@ -8,9 +8,13 @@ auth:
   type: token
   token: <CREDENTIAL>
 
-### cascading ###
-downstreams:
-  zipper-asia:
+### cascading mesh ###
+mesh:
+  zipper-sgp:
+    host: 1.1.1.1
+    port: 9000
+    credential: "token: <CREDENTIAL>"
+  zipper-aus:
     host: 2.2.2.2
     port: 9000
     credential: "token: <CREDENTIAL>"
@@ -21,4 +25,4 @@ downstreams:
   zipper-deu:
     host: 4.4.4.4
     port: 9000
-    auth: "token: <CREDENTIAL>"
+    auth: "token: <CREDENTIAL>"

zipper.go

@@ -43,7 +43,7 @@ func RunZipper(ctx context.Context, configPath string) error {
 		}
 	}
 
-	zipper, err := NewZipper(conf.Name, router.Default(), conf.Downstreams, options...)
+	zipper, err := NewZipper(conf.Name, router.Default(), conf.Mesh, options...)
 	if err != nil {
 		return err
 	}
@@ -53,7 +53,7 @@ func RunZipper(ctx context.Context, configPath string) error {
 }
 
 // NewZipper returns a zipper.
-func NewZipper(name string, router router.Router, meshConfig map[string]config.Downstream, options ...ZipperOption) (Zipper, error) {
+func NewZipper(name string, router router.Router, meshConfig map[string]config.Mesh, options ...ZipperOption) (Zipper, error) {
 	opts := &zipperOptions{}
 
 	for _, o := range options {
@@ -63,19 +63,22 @@ func NewZipper(name string, router router.Router, meshConfig map[string]config.D
 	server := core.NewServer(name, opts.serverOption...)
 
 	// add downstreams to server.
-	for downstreamName, meshConf := range meshConfig {
+	for meshName, meshConf := range meshConfig {
+		if meshName == "" || meshName == name {
+			continue
+		}
 		addr := fmt.Sprintf("%s:%d", meshConf.Host, meshConf.Port)
 
 		clientOptions := []core.ClientOption{
 			core.WithCredential(meshConf.Credential),
 			core.WithNonBlockWrite(),
 			core.WithReConnect(),
-			core.WithLogger(server.Logger().With("downstream_name", downstreamName, "downstream_addr", addr)),
+			core.WithLogger(server.Logger().With("downstream_name", meshName, "downstream_addr", addr)),
 		}
 		clientOptions = append(clientOptions, opts.clientOption...)
 
 		downstream := &downstream{
-			localName: downstreamName,
+			localName: meshName,
 			client:    core.NewClient(name, addr, core.ClientTypeUpstreamZipper, clientOptions...),
 		}