Repository: numaproj/numaflow
Pull request #70 (2022-06-23 21:07:05)

feat: define buffer limits on edges
Signed-off-by: Derek Wang <whynowy@gmail.com>

`bufferMaxLength` and `bufferUsageLimit` are the properties for a buffer instead of a vertex, which is currently incorrectly defined in Vertex level.

This PR removes these 2 properties from Vertex Limits and add them into Edge Limits. In Pipeline level Limits definition, they remain in the same place.

<!--

Before you push your changes:

* Run `make pre-push -B` to fix codegen and lint problems.

Then, you MUST:

* Sign-off your commit (otherwise the DCO check will fail).
* Use [a conventional commit message](https://www.conventionalcommits.org/en/v1.0.0/) (otherwise the commit message check will fail).

If you did not do this, reset all your commit and replace them with a single commit:

```
git reset HEAD~1 ;# change 1 to how many commits you made
git commit --signoff -m 'feat: my feat. Fixes #1234'
```

When creating your PR: 

* Make sure that "Fixes #" or "Closes #" is in both the PR title (for release notes) and description (to automatically link and close the issue).
* Say how you tested your changes. If you changed the UI, attach screenshots.
* Set your PR as a draft initially.
* Your PR needs to pass the required checks before it can be approved. 
* Once required tests have passed, mark your PR "Ready for review".

If changes were requested, once you've made them, you MUST dismiss the review to get it reviewed again.

-->



Patch:
diff --git a/.github/pull_request_template.md b/.github/pull_request_template.md
index 611aaab6..9a875103 100644
--- a/.github/pull_request_template.md
+++ b/.github/pull_request_template.md
@@ -1,9 +1,5 @@
 Explain what this PR does.
 
-Fixes #TODO
-or
-Closes #TODO
-
 <!--
 
 Before you push your changes:
diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml
index 21b2a93d..01d970f7 100644
--- a/.github/workflows/ci.yaml
+++ b/.github/workflows/ci.yaml
@@ -121,7 +121,7 @@ jobs:
       max-parallel: 4
       matrix:
         driver: [jetstream]
-        case: [e2e, kafka-e2e]
+        case: [e2e, kafka-e2e, http-e2e]
     steps:
       - name: Checkout code
         uses: actions/checkout@v3
diff --git a/Makefile b/Makefile
index 848ecb5b..7077467b 100644
--- a/Makefile
+++ b/Makefile
@@ -95,6 +95,7 @@ test-code:
 
 test-e2e:
 test-kafka-e2e:
+test-http-e2e:
 test-%: 
 	$(MAKE) image e2eapi-image
 	kubectl -n numaflow-system delete po -lapp.kubernetes.io/component=controller-manager,app.kubernetes.io/part-of=numaflow
diff --git a/cmd/commands/controller.go b/cmd/commands/controller.go
index 93f5f2c3..a3887358 100644
--- a/cmd/commands/controller.go
+++ b/cmd/commands/controller.go
@@ -9,7 +9,7 @@ func NewControllerCommand() *cobra.Command {
 
 	command := &cobra.Command{
 		Use:   "controller",
-		Short: "Start a numaflow controller",
+		Short: "Start a NumaFlow controller",
 		Run: func(cmd *cobra.Command, args []string) {
 			ctrlcmd.Start()
 		},
diff --git a/cmd/commands/root.go b/cmd/commands/root.go
index 6c82b601..5d7499d6 100644
--- a/cmd/commands/root.go
+++ b/cmd/commands/root.go
@@ -26,4 +26,5 @@ func init() {
 	rootCmd.AddCommand(NewISBSvcBufferValidateCommand())
 	rootCmd.AddCommand(NewBuiltinUDFCommand())
 	rootCmd.AddCommand(NewDaemonServerCommand())
+	rootCmd.AddCommand(NewServerCommand())
 }
diff --git a/cmd/commands/server.go b/cmd/commands/server.go
new file mode 100644
index 00000000..3620d413
--- /dev/null
+++ b/cmd/commands/server.go
@@ -0,0 +1,17 @@
+package commands
+
+import (
+	svrcmd "github.com/numaproj/numaflow/server/cmd"
+	"github.com/spf13/cobra"
+)
+
+func NewServerCommand() *cobra.Command {
+	command := &cobra.Command{
+		Use:   "server",
+		Short: "Start a NumaFlow server",
+		Run: func(cmd *cobra.Command, args []string) {
+			svrcmd.Start()
+		},
+	}
+	return command
+}
diff --git a/config/base/crds/numaflow.numaproj.io_pipelines.yaml b/config/base/crds/numaflow.numaproj.io_pipelines.yaml
index fd363084..e5b90fe0 100644
--- a/config/base/crds/numaflow.numaproj.io_pipelines.yaml
+++ b/config/base/crds/numaflow.numaproj.io_pipelines.yaml
@@ -59,6 +59,26 @@ spec:
                       type: object
                     from:
                       type: string
+                    limits:
+                      description: Limits define the limitations such as buffer read
+                        batch size for the edge, will override pipeline level settings
+                      properties:
+                        bufferMaxLength:
+                          description: BufferMaxLength is used to define the max length
+                            of a buffer. It overrides the settings from pipeline limits.
+                            Only meaningful for UDF and Source vertice as only they
+                            do buffer write.
+                          format: int64
+                          type: integer
+                        bufferUsageLimit:
+                          description: BufferUsageLimit is used to define the pencentage
+                            of the buffer usage limit, a valid value should be less
+                            than 100, for example, 85. It overrides the settings from
+                            pipeline limits. Only meaningful for UDF and Source vertice
+                            as only they do buffer write.
+                          format: int32
+                          type: integer
+                      type: object
                     to:
                       type: string
                   required:
@@ -1369,21 +1389,6 @@ spec:
                         batch size for all the vertices of a pipleine, will override
                         pipeline level settings
                       properties:
-                        bufferMaxLength:
-                          description: BufferMaxLength is used to define the max length
-                            of a buffer. It overrides the settings from pipeline limits.
-                            Only meaningful for UDF and Source vertice as only they
-                            do buffer write.
-                          format: int64
-                          type: integer
-                        bufferUsageLimit:
-                          description: BufferUsageLimit is used to define the pencentage
-                            of the buffer usage limit, a valid value should be less
-                            than 100, for example, 85. It overrides the settings from
-                            pipeline limits. Only meaningful for UDF and Source vertice
-                            as only they do buffer write.
-                          format: int32
-                          type: integer
                         readBatchSize:
                           description: Read batch size
                           format: int64
diff --git a/config/base/crds/numaflow.numaproj.io_vertices.yaml b/config/base/crds/numaflow.numaproj.io_vertices.yaml
index 9b87e4af..e70a6727 100644
--- a/config/base/crds/numaflow.numaproj.io_vertices.yaml
+++ b/config/base/crds/numaflow.numaproj.io_vertices.yaml
@@ -1191,9 +1191,48 @@ spec:
                         type: object
                     type: object
                 type: object
-              fromVertices:
+              fromEdges:
                 items:
-                  type: string
+                  properties:
+                    conditions:
+                      description: Conditional forwarding, only allowed when "From"
+                        is a Sink or UDF
+                      properties:
+                        keyIn:
+                          items:
+                            type: string
+                          type: array
+                      required:
+                      - keyIn
+                      type: object
+                    from:
+                      type: string
+                    limits:
+                      description: Limits define the limitations such as buffer read
+                        batch size for the edge, will override pipeline level settings
+                      properties:
+                        bufferMaxLength:
+                          description: BufferMaxLength is used to define the max length
+                            of a buffer. It overrides the settings from pipeline limits.
+                            Only meaningful for UDF and Source vertice as only they
+                            do buffer write.
+                          format: int64
+                          type: integer
+                        bufferUsageLimit:
+                          description: BufferUsageLimit is used to define the pencentage
+                            of the buffer usage limit, a valid value should be less
+                            than 100, for example, 85. It overrides the settings from
+                            pipeline limits. Only meaningful for UDF and Source vertice
+                            as only they do buffer write.
+                          format: int32
+                          type: integer
+                      type: object
+                    to:
+                      type: string
+                  required:
+                  - from
+                  - to
+                  type: object
                 type: array
               imagePullSecrets:
                 description: 'ImagePullSecrets is an optional list of references to
@@ -1219,21 +1258,6 @@ spec:
                   size for all the vertices of a pipleine, will override pipeline
                   level settings
                 properties:
-                  bufferMaxLength:
-                    description: BufferMaxLength is used to define the max length
-                      of a buffer. It overrides the settings from pipeline limits.
-                      Only meaningful for UDF and Source vertice as only they do buffer
-                      write.
-                    format: int64
-                    type: integer
-                  bufferUsageLimit:
-                    description: BufferUsageLimit is used to define the pencentage
-                      of the buffer usage limit, a valid value should be less than
-                      100, for example, 85. It overrides the settings from pipeline
-                      limits. Only meaningful for UDF and Source vertice as only they
-                      do buffer write.
-                    format: int32
-                    type: integer
                   readBatchSize:
                     description: Read batch size
                     format: int64
@@ -1889,10 +1913,12 @@ spec:
                     - topic
                     type: object
                 type: object
-              toVertices:
+              toEdges:
                 items:
                   properties:
                     conditions:
+                      description: Conditional forwarding, only allowed when "From"
+                        is a Sink or UDF
                       properties:
                         keyIn:
                           items:
@@ -1901,10 +1927,33 @@ spec:
                       required:
                       - keyIn
                       type: object
-                    name:
+                    from:
+                      type: string
+                    limits:
+                      description: Limits define the limitations such as buffer read
+                        batch size for the edge, will override pipeline level settings
+                      properties:
+                        bufferMaxLength:
+                          description: BufferMaxLength is used to define the max length
+                            of a buffer. It overrides the settings from pipeline limits.
+                            Only meaningful for UDF and Source vertice as only they
+                            do buffer write.
+                          format: int64
+                          type: integer
+                        bufferUsageLimit:
+                          description: BufferUsageLimit is used to define the pencentage
+                            of the buffer usage limit, a valid value should be less
+                            than 100, for example, 85. It overrides the settings from
+                            pipeline limits. Only meaningful for UDF and Source vertice
+                            as only they do buffer write.
+                          format: int32
+                          type: integer
+                      type: object
+                    to:
                       type: string
                   required:
-                  - name
+                  - from
+                  - to
                   type: object
                 type: array
               tolerations:
diff --git a/config/install.yaml b/config/install.yaml
index ac9fcb49..ce7b1da4 100644
--- a/config/install.yaml
+++ b/config/install.yaml
@@ -4785,6 +4785,26 @@ spec:
                       type: object
                     from:
                       type: string
+                    limits:
+                      description: Limits define the limitations such as buffer read
+                        batch size for the edge, will override pipeline level settings
+                      properties:
+                        bufferMaxLength:
+                          description: BufferMaxLength is used to define the max length
+                            of a buffer. It overrides the settings from pipeline limits.
+                            Only meaningful for UDF and Source vertice as only they
+                            do buffer write.
+                          format: int64
+                          type: integer
+                        bufferUsageLimit:
+                          description: BufferUsageLimit is used to define the pencentage
+                            of the buffer usage limit, a valid value should be less
+                            than 100, for example, 85. It overrides the settings from
+                            pipeline limits. Only meaningful for UDF and Source vertice
+                            as only they do buffer write.
+                          format: int32
+                          type: integer
+                      type: object
                     to:
                       type: string
                   required:
@@ -6095,21 +6115,6 @@ spec:
                         batch size for all the vertices of a pipleine, will override
                         pipeline level settings
                       properties:
-                        bufferMaxLength:
-                          description: BufferMaxLength is used to define the max length
-                            of a buffer. It overrides the settings from pipeline limits.
-                            Only meaningful for UDF and Source vertice as only they
-                            do buffer write.
-                          format: int64
-                          type: integer
-                        bufferUsageLimit:
-                          description: BufferUsageLimit is used to define the pencentage
-                            of the buffer usage limit, a valid value should be less
-                            than 100, for example, 85. It overrides the settings from
-                            pipeline limits. Only meaningful for UDF and Source vertice
-                            as only they do buffer write.
-                          format: int32
-                          type: integer
                         readBatchSize:
                           description: Read batch size
                           format: int64
@@ -9927,9 +9932,48 @@ spec:
                         type: object
                     type: object
                 type: object
-              fromVertices:
+              fromEdges:
                 items:
-                  type: string
+                  properties:
+                    conditions:
+                      description: Conditional forwarding, only allowed when "From"
+                        is a Sink or UDF
+                      properties:
+                        keyIn:
+                          items:
+                            type: string
+                          type: array
+                      required:
+                      - keyIn
+                      type: object
+                    from:
+                      type: string
+                    limits:
+                      description: Limits define the limitations such as buffer read
+                        batch size for the edge, will override pipeline level settings
+                      properties:
+                        bufferMaxLength:
+                          description: BufferMaxLength is used to define the max length
+                            of a buffer. It overrides the settings from pipeline limits.
+                            Only meaningful for UDF and Source vertice as only they
+                            do buffer write.
+                          format: int64
+                          type: integer
+                        bufferUsageLimit:
+                          description: BufferUsageLimit is used to define the pencentage
+                            of the buffer usage limit, a valid value should be less
+                            than 100, for example, 85. It overrides the settings from
+                            pipeline limits. Only meaningful for UDF and Source vertice
+                            as only they do buffer write.
+                          format: int32
+                          type: integer
+                      type: object
+                    to:
+                      type: string
+                  required:
+                  - from
+                  - to
+                  type: object
                 type: array
               imagePullSecrets:
                 description: 'ImagePullSecrets is an optional list of references to
@@ -9955,21 +9999,6 @@ spec:
                   size for all the vertices of a pipleine, will override pipeline
                   level settings
                 properties:
-                  bufferMaxLength:
-                    description: BufferMaxLength is used to define the max length
-                      of a buffer. It overrides the settings from pipeline limits.
-                      Only meaningful for UDF and Source vertice as only they do buffer
-                      write.
-                    format: int64
-                    type: integer
-                  bufferUsageLimit:
-                    description: BufferUsageLimit is used to define the pencentage
-                      of the buffer usage limit, a valid value should be less than
-                      100, for example, 85. It overrides the settings from pipeline
-                      limits. Only meaningful for UDF and Source vertice as only they
-                      do buffer write.
-                    format: int32
-                    type: integer
                   readBatchSize:
                     description: Read batch size
                     format: int64
@@ -10625,10 +10654,12 @@ spec:
                     - topic
                     type: object
                 type: object
-              toVertices:
+              toEdges:
                 items:
                   properties:
                     conditions:
+                      description: Conditional forwarding, only allowed when "From"
+                        is a Sink or UDF
                       properties:
                         keyIn:
                           items:
@@ -10637,10 +10668,33 @@ spec:
                       required:
                       - keyIn
                       type: object
-                    name:
+                    from:
+                      type: string
+                    limits:
+                      description: Limits define the limitations such as buffer read
+                        batch size for the edge, will override pipeline level settings
+                      properties:
+                        bufferMaxLength:
+                          description: BufferMaxLength is used to define the max length
+                            of a buffer. It overrides the settings from pipeline limits.
+                            Only meaningful for UDF and Source vertice as only they
+                            do buffer write.
+                          format: int64
+                          type: integer
+                        bufferUsageLimit:
+                          description: BufferUsageLimit is used to define the pencentage
+                            of the buffer usage limit, a valid value should be less
+                            than 100, for example, 85. It overrides the settings from
+                            pipeline limits. Only meaningful for UDF and Source vertice
+                            as only they do buffer write.
+                          format: int32
+                          type: integer
+                      type: object
+                    to:
                       type: string
                   required:
-                  - name
+                  - from
+                  - to
                   type: object
                 type: array
               tolerations:
diff --git a/controllers/cmd/start.go b/controllers/cmd/start.go
index cd684021..a4313518 100644
--- a/controllers/cmd/start.go
+++ b/controllers/cmd/start.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package controllers
+package cmd
 
 import (
 	"reflect"
diff --git a/controllers/pipeline/controller.go b/controllers/pipeline/controller.go
index 46573ccc..63fadfe8 100644
--- a/controllers/pipeline/controller.go
+++ b/controllers/pipeline/controller.go
@@ -411,16 +411,10 @@ func buildVertices(pl *dfv1.Pipeline) map[string]dfv1.Vertex {
 			dfv1.KeyPipelineName: pl.Name,
 			dfv1.KeyVertexName:   v.Name,
 		}
-		fromVertexNames := []string{}
-		toVertices := []dfv1.ToVertex{}
-		for _, e := range pl.GetFromEdges(v.Name) {
-			fromVertexNames = append(fromVertexNames, e.From)
-		}
-		for _, e := range pl.GetToEdges(v.Name) {
-			toVertices = append(toVertices, dfv1.ToVertex{Name: e.To, Conditions: e.Conditions})
-		}
+		fromEdges := copyEdgeLimits(pl, pl.GetFromEdges(v.Name))
+		toEdges := copyEdgeLimits(pl, pl.GetToEdges(v.Name))
 		vCopy := v.DeepCopy()
-		copyLimits(pl, vCopy)
+		copyVertexLimits(pl, vCopy)
 		replicas := int32(1)
 		if pl.Status.Phase == dfv1.PipelinePhasePaused {
 			replicas = int32(0)
@@ -437,8 +431,8 @@ func buildVertices(pl *dfv1.Pipeline) map[string]dfv1.Vertex {
 			AbstractVertex:             *vCopy,
 			PipelineName:               pl.Name,
 			InterStepBufferServiceName: pl.Spec.InterStepBufferServiceName,
-			FromVertices:               fromVertexNames,
-			ToVertices:                 toVertices,
+			FromEdges:                  fromEdges,
+			ToEdges:                    toEdges,
 			Replicas:                   &replicas,
 		}
 		hash := sharedutil.MustHash(spec.WithOutReplicas())
@@ -461,7 +455,7 @@ func buildVertices(pl *dfv1.Pipeline) map[string]dfv1.Vertex {
 	return result
 }
 
-func copyLimits(pl *dfv1.Pipeline, v *dfv1.AbstractVertex) {
+func copyVertexLimits(pl *dfv1.Pipeline, v *dfv1.AbstractVertex) {
 	if pl.Spec.Limits == nil {
 		return
 	}
@@ -474,12 +468,26 @@ func copyLimits(pl *dfv1.Pipeline, v *dfv1.AbstractVertex) {
 	if v.UDF != nil && v.Limits.UDFWorkers == nil {
 		v.Limits.UDFWorkers = pl.Spec.Limits.UDFWorkers
 	}
-	if v.Sink == nil && v.Limits.BufferMaxLength == nil {
-		v.Limits.BufferMaxLength = pl.Spec.Limits.BufferMaxLength
+}
+
+func copyEdgeLimits(pl *dfv1.Pipeline, edges []dfv1.Edge) []dfv1.Edge {
+	if pl.Spec.Limits == nil {
+		return edges
 	}
-	if v.Sink == nil && v.Limits.BufferUsageLimit == nil {
-		v.Limits.BufferUsageLimit = pl.Spec.Limits.BufferUsageLimit
+	result := []dfv1.Edge{}
+	for _, e := range edges {
+		if e.Limits == nil {
+			e.Limits = &dfv1.EdgeLimits{}
+		}
+		if e.Limits.BufferMaxLength == nil {
+			e.Limits.BufferMaxLength = pl.Spec.Limits.BufferMaxLength
+		}
+		if e.Limits.BufferUsageLimit == nil {
+			e.Limits.BufferUsageLimit = pl.Spec.Limits.BufferUsageLimit
+		}
+		result = append(result, e)
 	}
+	return result
 }
 
 func buildISBBatchJob(pl *dfv1.Pipeline, image string, isbSvcConfig dfv1.BufferServiceConfig, subCommand string, args []string, jobType string) *batchv1.Job {
diff --git a/docs/APIs.md b/docs/APIs.md
index 44ca972e..822d7f99 100644
--- a/docs/APIs.md
+++ b/docs/APIs.md
@@ -528,7 +528,8 @@ Edge
 </h3>
 <p>
 (<em>Appears on:</em>
-<a href="#numaflow.numaproj.io/v1alpha1.PipelineSpec">PipelineSpec</a>)
+<a href="#numaflow.numaproj.io/v1alpha1.PipelineSpec">PipelineSpec</a>,
+<a href="#numaflow.numaproj.io/v1alpha1.VertexSpec">VertexSpec</a>)
 </p>
 <p>
 </p>
@@ -571,6 +572,70 @@ Conditional forwarding, only allowed when “From” is a Sink or UDF
 </p>
 </td>
 </tr>
+<tr>
+<td>
+<code>limits</code></br> <em>
+<a href="#numaflow.numaproj.io/v1alpha1.EdgeLimits"> EdgeLimits </a>
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>
+Limits define the limitations such as buffer read batch size for the
+edge, will override pipeline level settings
+</p>
+</td>
+</tr>
+</tbody>
+</table>
+<h3 id="numaflow.numaproj.io/v1alpha1.EdgeLimits">
+EdgeLimits
+</h3>
+<p>
+(<em>Appears on:</em>
+<a href="#numaflow.numaproj.io/v1alpha1.Edge">Edge</a>)
+</p>
+<p>
+</p>
+<table>
+<thead>
+<tr>
+<th>
+Field
+</th>
+<th>
+Description
+</th>
+</tr>
+</thead>
+<tbody>
+<tr>
+<td>
+<code>bufferMaxLength</code></br> <em> uint64 </em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>
+BufferMaxLength is used to define the max length of a buffer. It
+overrides the settings from pipeline limits. Only meaningful for UDF and
+Source vertice as only they do buffer write.
+</p>
+</td>
+</tr>
+<tr>
+<td>
+<code>bufferUsageLimit</code></br> <em> uint32 </em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>
+BufferUsageLimit is used to define the pencentage of the buffer usage
+limit, a valid value should be less than 100, for example, 85. It
+overrides the settings from pipeline limits. Only meaningful for UDF and
+Source vertice as only they do buffer write.
+</p>
+</td>
+</tr>
 </tbody>
 </table>
 <h3 id="numaflow.numaproj.io/v1alpha1.ForwardConditions">
@@ -578,8 +643,7 @@ ForwardConditions
 </h3>
 <p>
 (<em>Appears on:</em>
-<a href="#numaflow.numaproj.io/v1alpha1.Edge">Edge</a>,
-<a href="#numaflow.numaproj.io/v1alpha1.ToVertex">ToVertex</a>)
+<a href="#numaflow.numaproj.io/v1alpha1.Edge">Edge</a>)
 </p>
 <p>
 </p>
@@ -3133,46 +3197,6 @@ KeySecret refers to the secret that contains the key
 </tr>
 </tbody>
 </table>
-<h3 id="numaflow.numaproj.io/v1alpha1.ToVertex">
-ToVertex
-</h3>
-<p>
-(<em>Appears on:</em>
-<a href="#numaflow.numaproj.io/v1alpha1.VertexSpec">VertexSpec</a>)
-</p>
-<p>
-</p>
-<table>
-<thead>
-<tr>
-<th>
-Field
-</th>
-<th>
-Description
-</th>
-</tr>
-</thead>
-<tbody>
-<tr>
-<td>
-<code>name</code></br> <em> string </em>
-</td>
-<td>
-</td>
-</tr>
-<tr>
-<td>
-<code>conditions</code></br> <em>
-<a href="#numaflow.numaproj.io/v1alpha1.ForwardConditions">
-ForwardConditions </a> </em>
-</td>
-<td>
-<em>(Optional)</em>
-</td>
-</tr>
-</tbody>
-</table>
 <h3 id="numaflow.numaproj.io/v1alpha1.UDF">
 UDF
 </h3>
@@ -3319,7 +3343,8 @@ Refer to the Kubernetes API documentation for the fields of the
 </tr>
 <tr>
 <td>
-<code>fromVertices</code></br> <em> \[\]string </em>
+<code>fromEdges</code></br> <em>
+<a href="#numaflow.numaproj.io/v1alpha1.Edge"> \[\]Edge </a> </em>
 </td>
 <td>
 <em>(Optional)</em>
@@ -3327,9 +3352,8 @@ Refer to the Kubernetes API documentation for the fields of the
 </tr>
 <tr>
 <td>
-<code>toVertices</code></br> <em>
-<a href="#numaflow.numaproj.io/v1alpha1.ToVertex"> \[\]ToVertex </a>
-</em>
+<code>toEdges</code></br> <em>
+<a href="#numaflow.numaproj.io/v1alpha1.Edge"> \[\]Edge </a> </em>
 </td>
 <td>
 <em>(Optional)</em>
@@ -3395,33 +3419,6 @@ overrides the setting in pipeline limits.
 </p>
 </td>
 </tr>
-<tr>
-<td>
-<code>bufferMaxLength</code></br> <em> uint64 </em>
-</td>
-<td>
-<em>(Optional)</em>
-<p>
-BufferMaxLength is used to define the max length of a buffer. It
-overrides the settings from pipeline limits. Only meaningful for UDF and
-Source vertice as only they do buffer write.
-</p>
-</td>
-</tr>
-<tr>
-<td>
-<code>bufferUsageLimit</code></br> <em> uint32 </em>
-</td>
-<td>
-<em>(Optional)</em>
-<p>
-BufferUsageLimit is used to define the pencentage of the buffer usage
-limit, a valid value should be less than 100, for example, 85. It
-overrides the settings from pipeline limits. Only meaningful for UDF and
-Source vertice as only they do buffer write.
-</p>
-</td>
-</tr>
 </tbody>
 </table>
 <h3 id="numaflow.numaproj.io/v1alpha1.VertexPhase">
@@ -3492,7 +3489,8 @@ Description
 </tr>
 <tr>
 <td>
-<code>fromVertices</code></br> <em> \[\]string </em>
+<code>fromEdges</code></br> <em>
+<a href="#numaflow.numaproj.io/v1alpha1.Edge"> \[\]Edge </a> </em>
 </td>
 <td>
 <em>(Optional)</em>
@@ -3500,9 +3498,8 @@ Description
 </tr>
 <tr>
 <td>
-<code>toVertices</code></br> <em>
-<a href="#numaflow.numaproj.io/v1alpha1.ToVertex"> \[\]ToVertex </a>
-</em>
+<code>toEdges</code></br> <em>
+<a href="#numaflow.numaproj.io/v1alpha1.Edge"> \[\]Edge </a> </em>
 </td>
 <td>
 <em>(Optional)</em>
diff --git a/pkg/apis/numaflow/v1alpha1/edge_types.go b/pkg/apis/numaflow/v1alpha1/edge_types.go
new file mode 100644
index 00000000..95342a6c
--- /dev/null
+++ b/pkg/apis/numaflow/v1alpha1/edge_types.go
@@ -0,0 +1,45 @@
+/*
+
+
+Licensed 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 v1alpha1
+
+type Edge struct {
+	From string `json:"from" protobuf:"bytes,1,opt,name=from"`
+	To   string `json:"to" protobuf:"bytes,2,opt,name=to"`
+	// Conditional forwarding, only allowed when "From" is a Sink or UDF
+	// +optional
+	Conditions *ForwardConditions `json:"conditions" protobuf:"bytes,3,opt,name=conditions"`
+	// Limits define the limitations such as buffer read batch size for the edge, will override pipeline level settings
+	// +optional
+	Limits *EdgeLimits `json:"limits,omitempty" protobuf:"bytes,4,opt,name=limits"`
+}
+
+type ForwardConditions struct {
+	KeyIn []string `json:"keyIn" protobuf:"bytes,1,rep,name=keyIn"`
+}
+
+type EdgeLimits struct {
+	// BufferMaxLength is used to define the max length of a buffer.
+	// It overrides the settings from pipeline limits.
+	// Only meaningful for UDF and Source vertice as only they do buffer write.
+	// +optional
+	BufferMaxLength *uint64 `json:"bufferMaxLength,omitempty" protobuf:"varint,1,opt,name=bufferMaxLength"`
+	// BufferUsageLimit is used to define the pencentage of the buffer usage limit, a valid value should be less than 100, for example, 85.
+	// It overrides the settings from pipeline limits.
+	// Only meaningful for UDF and Source vertice as only they do buffer write.
+	// +optional
+	BufferUsageLimit *uint32 `json:"bufferUsageLimit,omitempty" protobuf:"varint,2,opt,name=bufferUsageLimit"`
+}
diff --git a/pkg/apis/numaflow/v1alpha1/generated.pb.go b/pkg/apis/numaflow/v1alpha1/generated.pb.go
index e0bbad81..f815666f 100644
--- a/pkg/apis/numaflow/v1alpha1/generated.pb.go
+++ b/pkg/apis/numaflow/v1alpha1/generated.pb.go
@@ -244,10 +244,38 @@ func (m *Edge) XXX_DiscardUnknown() {
 
 var xxx_messageInfo_Edge proto.InternalMessageInfo
 
+func (m *EdgeLimits) Reset()      { *m = EdgeLimits{} }
+func (*EdgeLimits) ProtoMessage() {}
+func (*EdgeLimits) Descriptor() ([]byte, []int) {
+	return fileDescriptor_9d0d1b17d3865563, []int{7}
+}
+func (m *EdgeLimits) XXX_Unmarshal(b []byte) error {
+	return m.Unmarshal(b)
+}
+func (m *EdgeLimits) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	b = b[:cap(b)]
+	n, err := m.MarshalToSizedBuffer(b)
+	if err != nil {
+		return nil, err
+	}
+	return b[:n], nil
+}
+func (m *EdgeLimits) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_EdgeLimits.Merge(m, src)
+}
+func (m *EdgeLimits) XXX_Size() int {
+	return m.Size()
+}
+func (m *EdgeLimits) XXX_DiscardUnknown() {
+	xxx_messageInfo_EdgeLimits.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_EdgeLimits proto.InternalMessageInfo
+
 func (m *ForwardConditions) Reset()      { *m = ForwardConditions{} }
 func (*ForwardConditions) ProtoMessage() {}
 func (*ForwardConditions) Descriptor() ([]byte, []int) {
-	return fileDescriptor_9d0d1b17d3865563, []int{7}
+	return fileDescriptor_9d0d1b17d3865563, []int{8}
 }
 func (m *ForwardConditions) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -275,7 +303,7 @@ var xxx_messageInfo_ForwardConditions proto.InternalMessageInfo
 func (m *Function) Reset()      { *m = Function{} }
 func (*Function) ProtoMessage() {}
 func (*Function) Descriptor() ([]byte, []int) {
-	return fileDescriptor_9d0d1b17d3865563, []int{8}
+	return fileDescriptor_9d0d1b17d3865563, []int{9}
 }
 func (m *Function) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -303,7 +331,7 @@ var xxx_messageInfo_Function proto.InternalMessageInfo
 func (m *GeneratorSource) Reset()      { *m = GeneratorSource{} }
 func (*GeneratorSource) ProtoMessage() {}
 func (*GeneratorSource) Descriptor() ([]byte, []int) {
-	return fileDescriptor_9d0d1b17d3865563, []int{9}
+	return fileDescriptor_9d0d1b17d3865563, []int{10}
 }
 func (m *GeneratorSource) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -331,7 +359,7 @@ var xxx_messageInfo_GeneratorSource proto.InternalMessageInfo
 func (m *GetDaemonDeploymentReq) Reset()      { *m = GetDaemonDeploymentReq{} }
 func (*GetDaemonDeploymentReq) ProtoMessage() {}
 func (*GetDaemonDeploymentReq) Descriptor() ([]byte, []int) {
-	return fileDescriptor_9d0d1b17d3865563, []int{10}
+	return fileDescriptor_9d0d1b17d3865563, []int{11}
 }
 func (m *GetDaemonDeploymentReq) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -359,7 +387,7 @@ var xxx_messageInfo_GetDaemonDeploymentReq proto.InternalMessageInfo
 func (m *GetJetStreamServiceSpecReq) Reset()      { *m = GetJetStreamServiceSpecReq{} }
 func (*GetJetStreamServiceSpecReq) ProtoMessage() {}
 func (*GetJetStreamServiceSpecReq) Descriptor() ([]byte, []int) {
-	return fileDescriptor_9d0d1b17d3865563, []int{11}
+	return fileDescriptor_9d0d1b17d3865563, []int{12}
 }
 func (m *GetJetStreamServiceSpecReq) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -387,7 +415,7 @@ var xxx_messageInfo_GetJetStreamServiceSpecReq proto.InternalMessageInfo
 func (m *GetJetStreamStatefulSetSpecReq) Reset()      { *m = GetJetStreamStatefulSetSpecReq{} }
 func (*GetJetStreamStatefulSetSpecReq) ProtoMessage() {}
 func (*GetJetStreamStatefulSetSpecReq) Descriptor() ([]byte, []int) {
-	return fileDescriptor_9d0d1b17d3865563, []int{12}
+	return fileDescriptor_9d0d1b17d3865563, []int{13}
 }
 func (m *GetJetStreamStatefulSetSpecReq) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -415,7 +443,7 @@ var xxx_messageInfo_GetJetStreamStatefulSetSpecReq proto.InternalMessageInfo
 func (m *GetRedisServiceSpecReq) Reset()      { *m = GetRedisServiceSpecReq{} }
 func (*GetRedisServiceSpecReq) ProtoMessage() {}
 func (*GetRedisServiceSpecReq) Descriptor() ([]byte, []int) {
-	return fileDescriptor_9d0d1b17d3865563, []int{13}
+	return fileDescriptor_9d0d1b17d3865563, []int{14}
 }
 func (m *GetRedisServiceSpecReq) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -443,7 +471,7 @@ var xxx_messageInfo_GetRedisServiceSpecReq proto.InternalMessageInfo
 func (m *GetRedisStatefulSetSpecReq) Reset()      { *m = GetRedisStatefulSetSpecReq{} }
 func (*GetRedisStatefulSetSpecReq) ProtoMessage() {}
 func (*GetRedisStatefulSetSpecReq) Descriptor() ([]byte, []int) {
-	return fileDescriptor_9d0d1b17d3865563, []int{14}
+	return fileDescriptor_9d0d1b17d3865563, []int{15}
 }
 func (m *GetRedisStatefulSetSpecReq) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -471,7 +499,7 @@ var xxx_messageInfo_GetRedisStatefulSetSpecReq proto.InternalMessageInfo
 func (m *GetVertexPodSpecReq) Reset()      { *m = GetVertexPodSpecReq{} }
 func (*GetVertexPodSpecReq) ProtoMessage() {}
 func (*GetVertexPodSpecReq) Descriptor() ([]byte, []int) {
-	return fileDescriptor_9d0d1b17d3865563, []int{15}
+	return fileDescriptor_9d0d1b17d3865563, []int{16}
 }
 func (m *GetVertexPodSpecReq) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -499,7 +527,7 @@ var xxx_messageInfo_GetVertexPodSpecReq proto.InternalMessageInfo
 func (m *HTTPSource) Reset()      { *m = HTTPSource{} }
 func (*HTTPSource) ProtoMessage() {}
 func (*HTTPSource) Descriptor() ([]byte, []int) {
-	return fileDescriptor_9d0d1b17d3865563, []int{16}
+	return fileDescriptor_9d0d1b17d3865563, []int{17}
 }
 func (m *HTTPSource) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -527,7 +555,7 @@ var xxx_messageInfo_HTTPSource proto.InternalMessageInfo
 func (m *InterStepBufferService) Reset()      { *m = InterStepBufferService{} }
 func (*InterStepBufferService) ProtoMessage() {}
 func (*InterStepBufferService) Descriptor() ([]byte, []int) {
-	return fileDescriptor_9d0d1b17d3865563, []int{17}
+	return fileDescriptor_9d0d1b17d3865563, []int{18}
 }
 func (m *InterStepBufferService) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -555,7 +583,7 @@ var xxx_messageInfo_InterStepBufferService proto.InternalMessageInfo
 func (m *InterStepBufferServiceList) Reset()      { *m = InterStepBufferServiceList{} }
 func (*InterStepBufferServiceList) ProtoMessage() {}
 func (*InterStepBufferServiceList) Descriptor() ([]byte, []int) {
-	return fileDescriptor_9d0d1b17d3865563, []int{18}
+	return fileDescriptor_9d0d1b17d3865563, []int{19}
 }
 func (m *InterStepBufferServiceList) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -583,7 +611,7 @@ var xxx_messageInfo_InterStepBufferServiceList proto.InternalMessageInfo
 func (m *InterStepBufferServiceSpec) Reset()      { *m = InterStepBufferServiceSpec{} }
 func (*InterStepBufferServiceSpec) ProtoMessage() {}
 func (*InterStepBufferServiceSpec) Descriptor() ([]byte, []int) {
-	return fileDescriptor_9d0d1b17d3865563, []int{19}
+	return fileDescriptor_9d0d1b17d3865563, []int{20}
 }
 func (m *InterStepBufferServiceSpec) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -611,7 +639,7 @@ var xxx_messageInfo_InterStepBufferServiceSpec proto.InternalMessageInfo
 func (m *InterStepBufferServiceStatus) Reset()      { *m = InterStepBufferServiceStatus{} }
 func (*InterStepBufferServiceStatus) ProtoMessage() {}
 func (*InterStepBufferServiceStatus) Descriptor() ([]byte, []int) {
-	return fileDescriptor_9d0d1b17d3865563, []int{20}
+	return fileDescriptor_9d0d1b17d3865563, []int{21}
 }
 func (m *InterStepBufferServiceStatus) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -639,7 +667,7 @@ var xxx_messageInfo_InterStepBufferServiceStatus proto.InternalMessageInfo
 func (m *JetStreamBufferService) Reset()      { *m = JetStreamBufferService{} }
 func (*JetStreamBufferService) ProtoMessage() {}
 func (*JetStreamBufferService) Descriptor() ([]byte, []int) {
-	return fileDescriptor_9d0d1b17d3865563, []int{21}
+	return fileDescriptor_9d0d1b17d3865563, []int{22}
 }
 func (m *JetStreamBufferService) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -667,7 +695,7 @@ var xxx_messageInfo_JetStreamBufferService proto.InternalMessageInfo
 func (m *JetStreamConfig) Reset()      { *m = JetStreamConfig{} }
 func (*JetStreamConfig) ProtoMessage() {}
 func (*JetStreamConfig) Descriptor() ([]byte, []int) {
-	return fileDescriptor_9d0d1b17d3865563, []int{22}
+	return fileDescriptor_9d0d1b17d3865563, []int{23}
 }
 func (m *JetStreamConfig) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -695,7 +723,7 @@ var xxx_messageInfo_JetStreamConfig proto.InternalMessageInfo
 func (m *KafkaSink) Reset()      { *m = KafkaSink{} }
 func (*KafkaSink) ProtoMessage() {}
 func (*KafkaSink) Descriptor() ([]byte, []int) {
-	return fileDescriptor_9d0d1b17d3865563, []int{23}
+	return fileDescriptor_9d0d1b17d3865563, []int{24}
 }
 func (m *KafkaSink) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -723,7 +751,7 @@ var xxx_messageInfo_KafkaSink proto.InternalMessageInfo
 func (m *KafkaSource) Reset()      { *m = KafkaSource{} }
 func (*KafkaSource) ProtoMessage() {}
 func (*KafkaSource) Descriptor() ([]byte, []int) {
-	return fileDescriptor_9d0d1b17d3865563, []int{24}
+	return fileDescriptor_9d0d1b17d3865563, []int{25}
 }
 func (m *KafkaSource) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -751,7 +779,7 @@ var xxx_messageInfo_KafkaSource proto.InternalMessageInfo
 func (m *Lifecycle) Reset()      { *m = Lifecycle{} }
 func (*Lifecycle) ProtoMessage() {}
 func (*Lifecycle) Descriptor() ([]byte, []int) {
-	return fileDescriptor_9d0d1b17d3865563, []int{25}
+	return fileDescriptor_9d0d1b17d3865563, []int{26}
 }
 func (m *Lifecycle) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -779,7 +807,7 @@ var xxx_messageInfo_Lifecycle proto.InternalMessageInfo
 func (m *Log) Reset()      { *m = Log{} }
 func (*Log) ProtoMessage() {}
 func (*Log) Descriptor() ([]byte, []int) {
-	return fileDescriptor_9d0d1b17d3865563, []int{26}
+	return fileDescriptor_9d0d1b17d3865563, []int{27}
 }
 func (m *Log) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -807,7 +835,7 @@ var xxx_messageInfo_Log proto.InternalMessageInfo
 func (m *Metadata) Reset()      { *m = Metadata{} }
 func (*Metadata) ProtoMessage() {}
 func (*Metadata) Descriptor() ([]byte, []int) {
-	return fileDescriptor_9d0d1b17d3865563, []int{27}
+	return fileDescriptor_9d0d1b17d3865563, []int{28}
 }
 func (m *Metadata) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -835,7 +863,7 @@ var xxx_messageInfo_Metadata proto.InternalMessageInfo
 func (m *NATSAuth) Reset()      { *m = NATSAuth{} }
 func (*NATSAuth) ProtoMessage() {}
 func (*NATSAuth) Descriptor() ([]byte, []int) {
-	return fileDescriptor_9d0d1b17d3865563, []int{28}
+	return fileDescriptor_9d0d1b17d3865563, []int{29}
 }
 func (m *NATSAuth) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -863,7 +891,7 @@ var xxx_messageInfo_NATSAuth proto.InternalMessageInfo
 func (m *NativeRedis) Reset()      { *m = NativeRedis{} }
 func (*NativeRedis) ProtoMessage() {}
 func (*NativeRedis) Descriptor() ([]byte, []int) {
-	return fileDescriptor_9d0d1b17d3865563, []int{29}
+	return fileDescriptor_9d0d1b17d3865563, []int{30}
 }
 func (m *NativeRedis) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -891,7 +919,7 @@ var xxx_messageInfo_NativeRedis proto.InternalMessageInfo
 func (m *PersistenceStrategy) Reset()      { *m = PersistenceStrategy{} }
 func (*PersistenceStrategy) ProtoMessage() {}
 func (*PersistenceStrategy) Descriptor() ([]byte, []int) {
-	return fileDescriptor_9d0d1b17d3865563, []int{30}
+	return fileDescriptor_9d0d1b17d3865563, []int{31}
 }
 func (m *PersistenceStrategy) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -919,7 +947,7 @@ var xxx_messageInfo_PersistenceStrategy proto.InternalMessageInfo
 func (m *Pipeline) Reset()      { *m = Pipeline{} }
 func (*Pipeline) ProtoMessage() {}
 func (*Pipeline) Descriptor() ([]byte, []int) {
-	return fileDescriptor_9d0d1b17d3865563, []int{31}
+	return fileDescriptor_9d0d1b17d3865563, []int{32}
 }
 func (m *Pipeline) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -947,7 +975,7 @@ var xxx_messageInfo_Pipeline proto.InternalMessageInfo
 func (m *PipelineLimits) Reset()      { *m = PipelineLimits{} }
 func (*PipelineLimits) ProtoMessage() {}
 func (*PipelineLimits) Descriptor() ([]byte, []int) {
-	return fileDescriptor_9d0d1b17d3865563, []int{32}
+	return fileDescriptor_9d0d1b17d3865563, []int{33}
 }
 func (m *PipelineLimits) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -975,7 +1003,7 @@ var xxx_messageInfo_PipelineLimits proto.InternalMessageInfo
 func (m *PipelineList) Reset()      { *m = PipelineList{} }
 func (*PipelineList) ProtoMessage() {}
 func (*PipelineList) Descriptor() ([]byte, []int) {
-	return fileDescriptor_9d0d1b17d3865563, []int{33}
+	return fileDescriptor_9d0d1b17d3865563, []int{34}
 }
 func (m *PipelineList) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -1003,7 +1031,7 @@ var xxx_messageInfo_PipelineList proto.InternalMessageInfo
 func (m *PipelineSpec) Reset()      { *m = PipelineSpec{} }
 func (*PipelineSpec) ProtoMessage() {}
 func (*PipelineSpec) Descriptor() ([]byte, []int) {
-	return fileDescriptor_9d0d1b17d3865563, []int{34}
+	return fileDescriptor_9d0d1b17d3865563, []int{35}
 }
 func (m *PipelineSpec) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -1031,7 +1059,7 @@ var xxx_messageInfo_PipelineSpec proto.InternalMessageInfo
 func (m *PipelineStatus) Reset()      { *m = PipelineStatus{} }
 func (*PipelineStatus) ProtoMessage() {}
 func (*PipelineStatus) Descriptor() ([]byte, []int) {
-	return fileDescriptor_9d0d1b17d3865563, []int{35}
+	return fileDescriptor_9d0d1b17d3865563, []int{36}
 }
 func (m *PipelineStatus) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -1059,7 +1087,7 @@ var xxx_messageInfo_PipelineStatus proto.InternalMessageInfo
 func (m *RedisBuferService) Reset()      { *m = RedisBuferService{} }
 func (*RedisBuferService) ProtoMessage() {}
 func (*RedisBuferService) Descriptor() ([]byte, []int) {
-	return fileDescriptor_9d0d1b17d3865563, []int{36}
+	return fileDescriptor_9d0d1b17d3865563, []int{37}
 }
 func (m *RedisBuferService) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -1087,7 +1115,7 @@ var xxx_messageInfo_RedisBuferService proto.InternalMessageInfo
 func (m *RedisConfig) Reset()      { *m = RedisConfig{} }
 func (*RedisConfig) ProtoMessage() {}
 func (*RedisConfig) Descriptor() ([]byte, []int) {
-	return fileDescriptor_9d0d1b17d3865563, []int{37}
+	return fileDescriptor_9d0d1b17d3865563, []int{38}
 }
 func (m *RedisConfig) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -1115,7 +1143,7 @@ var xxx_messageInfo_RedisConfig proto.InternalMessageInfo
 func (m *RedisSettings) Reset()      { *m = RedisSettings{} }
 func (*RedisSettings) ProtoMessage() {}
 func (*RedisSettings) Descriptor() ([]byte, []int) {
-	return fileDescriptor_9d0d1b17d3865563, []int{38}
+	return fileDescriptor_9d0d1b17d3865563, []int{39}
 }
 func (m *RedisSettings) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -1143,7 +1171,7 @@ var xxx_messageInfo_RedisSettings proto.InternalMessageInfo
 func (m *Scale) Reset()      { *m = Scale{} }
 func (*Scale) ProtoMessage() {}
 func (*Scale) Descriptor() ([]byte, []int) {
-	return fileDescriptor_9d0d1b17d3865563, []int{39}
+	return fileDescriptor_9d0d1b17d3865563, []int{40}
 }
 func (m *Scale) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -1171,7 +1199,7 @@ var xxx_messageInfo_Scale proto.InternalMessageInfo
 func (m *Sink) Reset()      { *m = Sink{} }
 func (*Sink) ProtoMessage() {}
 func (*Sink) Descriptor() ([]byte, []int) {
-	return fileDescriptor_9d0d1b17d3865563, []int{40}
+	return fileDescriptor_9d0d1b17d3865563, []int{41}
 }
 func (m *Sink) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -1199,7 +1227,7 @@ var xxx_messageInfo_Sink proto.InternalMessageInfo
 func (m *Source) Reset()      { *m = Source{} }
 func (*Source) ProtoMessage() {}
 func (*Source) Descriptor() ([]byte, []int) {
-	return fileDescriptor_9d0d1b17d3865563, []int{41}
+	return fileDescriptor_9d0d1b17d3865563, []int{42}
 }
 func (m *Source) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -1227,7 +1255,7 @@ var xxx_messageInfo_Source proto.InternalMessageInfo
 func (m *Status) Reset()      { *m = Status{} }
 func (*Status) ProtoMessage() {}
 func (*Status) Descriptor() ([]byte, []int) {
-	return fileDescriptor_9d0d1b17d3865563, []int{42}
+	return fileDescriptor_9d0d1b17d3865563, []int{43}
 }
 func (m *Status) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -1255,7 +1283,7 @@ var xxx_messageInfo_Status proto.InternalMessageInfo
 func (m *TLS) Reset()      { *m = TLS{} }
 func (*TLS) ProtoMessage() {}
 func (*TLS) Descriptor() ([]byte, []int) {
-	return fileDescriptor_9d0d1b17d3865563, []int{43}
+	return fileDescriptor_9d0d1b17d3865563, []int{44}
 }
 func (m *TLS) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -1280,34 +1308,6 @@ func (m *TLS) XXX_DiscardUnknown() {
 
 var xxx_messageInfo_TLS proto.InternalMessageInfo
 
-func (m *ToVertex) Reset()      { *m = ToVertex{} }
-func (*ToVertex) ProtoMessage() {}
-func (*ToVertex) Descriptor() ([]byte, []int) {
-	return fileDescriptor_9d0d1b17d3865563, []int{44}
-}
-func (m *ToVertex) XXX_Unmarshal(b []byte) error {
-	return m.Unmarshal(b)
-}
-func (m *ToVertex) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
-	b = b[:cap(b)]
-	n, err := m.MarshalToSizedBuffer(b)
-	if err != nil {
-		return nil, err
-	}
-	return b[:n], nil
-}
-func (m *ToVertex) XXX_Merge(src proto.Message) {
-	xxx_messageInfo_ToVertex.Merge(m, src)
-}
-func (m *ToVertex) XXX_Size() int {
-	return m.Size()
-}
-func (m *ToVertex) XXX_DiscardUnknown() {
-	xxx_messageInfo_ToVertex.DiscardUnknown(m)
-}
-
-var xxx_messageInfo_ToVertex proto.InternalMessageInfo
-
 func (m *UDF) Reset()      { *m = UDF{} }
 func (*UDF) ProtoMessage() {}
 func (*UDF) Descriptor() ([]byte, []int) {
@@ -1541,6 +1541,7 @@ func init() {
 	proto.RegisterType((*Container)(nil), "github.com.numaproj.numaflow.pkg.apis.numaflow.v1alpha1.Container")
 	proto.RegisterType((*ContainerTemplate)(nil), "github.com.numaproj.numaflow.pkg.apis.numaflow.v1alpha1.ContainerTemplate")
 	proto.RegisterType((*Edge)(nil), "github.com.numaproj.numaflow.pkg.apis.numaflow.v1alpha1.Edge")
+	proto.RegisterType((*EdgeLimits)(nil), "github.com.numaproj.numaflow.pkg.apis.numaflow.v1alpha1.EdgeLimits")
 	proto.RegisterType((*ForwardConditions)(nil), "github.com.numaproj.numaflow.pkg.apis.numaflow.v1alpha1.ForwardConditions")
 	proto.RegisterType((*Function)(nil), "github.com.numaproj.numaflow.pkg.apis.numaflow.v1alpha1.Function")
 	proto.RegisterMapType((map[string]string)(nil), "github.com.numaproj.numaflow.pkg.apis.numaflow.v1alpha1.Function.KwargsEntry")
@@ -1587,7 +1588,6 @@ func init() {
 	proto.RegisterType((*Source)(nil), "github.com.numaproj.numaflow.pkg.apis.numaflow.v1alpha1.Source")
 	proto.RegisterType((*Status)(nil), "github.com.numaproj.numaflow.pkg.apis.numaflow.v1alpha1.Status")
 	proto.RegisterType((*TLS)(nil), "github.com.numaproj.numaflow.pkg.apis.numaflow.v1alpha1.TLS")
-	proto.RegisterType((*ToVertex)(nil), "github.com.numaproj.numaflow.pkg.apis.numaflow.v1alpha1.ToVertex")
 	proto.RegisterType((*UDF)(nil), "github.com.numaproj.numaflow.pkg.apis.numaflow.v1alpha1.UDF")
 	proto.RegisterType((*UDSink)(nil), "github.com.numaproj.numaflow.pkg.apis.numaflow.v1alpha1.UDSink")
 	proto.RegisterType((*Vertex)(nil), "github.com.numaproj.numaflow.pkg.apis.numaflow.v1alpha1.Vertex")
@@ -1603,285 +1603,286 @@ func init() {
 }
 
 var fileDescriptor_9d0d1b17d3865563 = []byte{
-	// 4446 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xec, 0x5c, 0x5d, 0x6c, 0x64, 0xc9,
-	0x55, 0xde, 0xdb, 0x7f, 0xee, 0x3e, 0x6d, 0x8f, 0xed, 0x9a, 0xd9, 0xa1, 0xc7, 0xcc, 0xba, 0x87,
-	0x5e, 0xed, 0x6a, 0x80, 0xa4, 0xcd, 0x9a, 0x09, 0x99, 0x40, 0x36, 0x13, 0xb7, 0xff, 0x76, 0x66,
-	0xda, 0xb3, 0xa6, 0xda, 0x9e, 0x61, 0xb3, 0x11, 0x43, 0xf9, 0x76, 0x75, 0xfb, 0x6e, 0xdf, 0xbe,
-	0xb7, 0x73, 0x6f, 0x5d, 0xcf, 0x78, 0x45, 0x04, 0x12, 0x0f, 0x0b, 0x0a, 0x52, 0x22, 0xf1, 0x82,
-	0x14, 0x81, 0x78, 0x88, 0x04, 0x3c, 0xf0, 0x82, 0xe0, 0x05, 0x14, 0x29, 0x4f, 0x68, 0x79, 0xdb,
-	0x07, 0x24, 0x16, 0x29, 0xb2, 0xb2, 0x46, 0xe2, 0x0d, 0x11, 0x94, 0x17, 0x34, 0x42, 0x02, 0xd5,
-	0xcf, 0xfd, 0xed, 0xdb, 0x1e, 0xbb, 0xdb, 0x9e, 0x3c, 0x64, 0xde, 0xfa, 0xd6, 0x39, 0xe7, 0x3b,
-	0x55, 0xa7, 0xaa, 0x4e, 0xd5, 0x39, 0x55, 0xd5, 0xb0, 0xd9, 0x35, 0xd8, 0xbe, 0xb7, 0x57, 0xd7,
-	0xed, 0xfe, 0x92, 0xe5, 0xf5, 0xc9, 0xc0, 0xb1, 0x3f, 0x10, 0x3f, 0x3a, 0xa6, 0xfd, 0x64, 0x69,
-	0xd0, 0xeb, 0x2e, 0x91, 0x81, 0xe1, 0x86, 0x25, 0x07, 0x6f, 0x11, 0x73, 0xb0, 0x4f, 0xde, 0x5a,
-	0xea, 0x52, 0x8b, 0x3a, 0x84, 0xd1, 0x76, 0x7d, 0xe0, 0xd8, 0xcc, 0x46, 0x5f, 0x0c, 0x81, 0xea,
-	0x3e, 0x50, 0xdd, 0x17, 0xab, 0x0f, 0x7a, 0xdd, 0x3a, 0x07, 0x0a, 0x4b, 0x7c, 0xa0, 0x85, 0xcf,
-	0x47, 0x6a, 0xd0, 0xb5, 0xbb, 0xf6, 0x92, 0xc0, 0xdb, 0xf3, 0x3a, 0xe2, 0x4b, 0x7c, 0x88, 0x5f,
-	0x52, 0xcf, 0x42, 0xad, 0x77, 0xdb, 0xad, 0x1b, 0x36, 0xaf, 0xd6, 0x92, 0x6e, 0x3b, 0x74, 0xe9,
-	0x60, 0xa8, 0x2e, 0x0b, 0xb7, 0x42, 0x9e, 0x3e, 0xd1, 0xf7, 0x0d, 0x8b, 0x3a, 0x87, 0x7e, 0x5b,
-	0x96, 0x1c, 0xea, 0xda, 0x9e, 0xa3, 0xd3, 0x33, 0x49, 0xb9, 0x4b, 0x7d, 0xca, 0x48, 0x9a, 0xae,
-	0xa5, 0x51, 0x52, 0x8e, 0x67, 0x31, 0xa3, 0x3f, 0xac, 0xe6, 0xd7, 0x9e, 0x27, 0xe0, 0xea, 0xfb,
-	0xb4, 0x4f, 0x92, 0x72, 0xb5, 0xff, 0x9a, 0x86, 0x4b, 0x2b, 0x7b, 0x2e, 0x73, 0x88, 0xce, 0x1e,
-	0x52, 0x87, 0xd1, 0xa7, 0xe8, 0x06, 0xe4, 0x2c, 0xd2, 0xa7, 0x15, 0xed, 0x86, 0x76, 0xb3, 0xd4,
-	0x98, 0xfe, 0xf8, 0xa8, 0xfa, 0xca, 0xf1, 0x51, 0x35, 0xf7, 0x80, 0xf4, 0x29, 0x16, 0x14, 0xa4,
-	0x43, 0x41, 0xb6, 0xb6, 0x92, 0xbd, 0xa1, 0xdd, 0x2c, 0x2f, 0xdf, 0xa9, 0x8f, 0xd9, 0x4d, 0xf5,
-	0x96, 0x80, 0x69, 0xc0, 0xf1, 0x51, 0xb5, 0x20, 0x7f, 0x63, 0x05, 0x8d, 0xde, 0x87, 0x9c, 0x6b,
-	0x58, 0xbd, 0x4a, 0x4e, 0xa8, 0x78, 0x7b, 0x7c, 0x15, 0x86, 0xd5, 0x6b, 0x14, 0x79, 0x0b, 0xf8,
-	0x2f, 0x2c, 0x40, 0xd1, 0xb7, 0x35, 0x98, 0xd7, 0x6d, 0x8b, 0x11, 0x6e, 0xa8, 0x1d, 0xda, 0x1f,
-	0x98, 0x84, 0xd1, 0x4a, 0x5e, 0xa8, 0xba, 0x37, 0xb6, 0xaa, 0xd5, 0x24, 0x62, 0xe3, 0xd5, 0xe3,
-	0xa3, 0xea, 0xfc, 0x50, 0x31, 0x1e, 0xd6, 0x8d, 0x1e, 0x41, 0xd6, 0x6b, 0x77, 0x2a, 0x05, 0x51,
-	0x85, 0x2f, 0x8f, 0x5d, 0x85, 0xdd, 0xb5, 0x8d, 0xc6, 0xd4, 0xf1, 0x51, 0x35, 0xbb, 0xbb, 0xb6,
-	0x81, 0x39, 0x22, 0xea, 0x41, 0x91, 0x8f, 0xb2, 0x36, 0x61, 0xa4, 0x32, 0x25, 0xd0, 0x57, 0xc6,
-	0x46, 0xdf, 0x52, 0x40, 0x8d, 0xe9, 0xe3, 0xa3, 0x6a, 0xd1, 0xff, 0xc2, 0x81, 0x02, 0xf4, 0x27,
-	0x1a, 0x4c, 0x5b, 0x76, 0x9b, 0xb6, 0xa8, 0x49, 0x75, 0x66, 0x3b, 0x95, 0xe2, 0x8d, 0xec, 0xcd,
-	0xf2, 0xf2, 0x7b, 0x63, 0x6b, 0x8c, 0x8f, 0xcd, 0xfa, 0x83, 0x08, 0xf6, 0xba, 0xc5, 0x9c, 0xc3,
-	0xc6, 0x15, 0x35, 0x3e, 0xa7, 0xa3, 0x24, 0x1c, 0xab, 0x04, 0xda, 0x85, 0x32, 0xb3, 0x4d, 0x3e,
-	0xee, 0x0d, 0xdb, 0x72, 0x2b, 0x25, 0x51, 0xa7, 0xc5, 0xba, 0x9c, 0x32, 0x5c, 0x73, 0x9d, 0xcf,
-	0xf9, 0xfa, 0xc1, 0x5b, 0xf5, 0x9d, 0x80, 0xad, 0x71, 0x59, 0x01, 0x97, 0xc3, 0x32, 0x17, 0x47,
-	0x71, 0x10, 0x85, 0x59, 0x97, 0xea, 0x9e, 0x63, 0xb0, 0x43, 0xde, 0xc5, 0xf4, 0x29, 0xab, 0x80,
-	0x30, 0xf0, 0x9b, 0x69, 0xd0, 0xdb, 0x76, 0xbb, 0x15, 0xe7, 0x6e, 0x5c, 0x3e, 0x3e, 0xaa, 0xce,
-	0x26, 0x0a, 0x71, 0x12, 0x13, 0x59, 0x30, 0x67, 0xf4, 0x49, 0x97, 0x6e, 0x7b, 0xa6, 0xd9, 0xa2,
-	0xba, 0x43, 0x99, 0x5b, 0x29, 0x8b, 0x26, 0xdc, 0x4c, 0xd3, 0xd3, 0xb4, 0x75, 0x62, 0xbe, 0xbb,
-	0xf7, 0x01, 0xd5, 0x19, 0xa6, 0x1d, 0xea, 0x50, 0x4b, 0xa7, 0x8d, 0x8a, 0x6a, 0xcc, 0xdc, 0xdd,
-	0x04, 0x12, 0x1e, 0xc2, 0x46, 0x9b, 0x30, 0x3f, 0x70, 0x0c, 0x5b, 0x54, 0xc1, 0x24, 0xae, 0xcb,
-	0x27, 0x7e, 0x65, 0x5a, 0x38, 0x83, 0x6b, 0x0a, 0x66, 0x7e, 0x3b, 0xc9, 0x80, 0x87, 0x65, 0xd0,
-	0x4d, 0x28, 0xfa, 0x85, 0x95, 0x99, 0x1b, 0xda, 0xcd, 0xbc, 0x1c, 0x36, 0xbe, 0x2c, 0x0e, 0xa8,
-	0x68, 0x03, 0x8a, 0xa4, 0xd3, 0x31, 0x2c, 0xce, 0x79, 0x49, 0x98, 0xf0, 0x7a, 0x5a, 0xd3, 0x56,
-	0x14, 0x8f, 0xc4, 0xf1, 0xbf, 0x70, 0x20, 0x8b, 0xee, 0x01, 0x72, 0xa9, 0x73, 0x60, 0xe8, 0x74,
-	0x45, 0xd7, 0x6d, 0xcf, 0x62, 0xa2, 0xee, 0xb3, 0xa2, 0xee, 0x0b, 0xaa, 0xee, 0xa8, 0x35, 0xc4,
-	0x81, 0x53, 0xa4, 0xd0, 0x3a, 0x4c, 0x1d, 0xd8, 0xa6, 0xd7, 0xa7, 0x6e, 0x65, 0x4e, 0x58, 0x7b,
-	0x21, 0xad, 0x4a, 0x0f, 0x05, 0x4b, 0x63, 0x56, 0x81, 0x4f, 0xc9, 0x6f, 0x17, 0xfb, 0xb2, 0xc8,
-	0x80, 0x82, 0x69, 0xf4, 0x0d, 0xe6, 0x56, 0xe6, 0x45, 0xc3, 0xd6, 0xc7, 0x9e, 0x0a, 0x72, 0x0a,
-	0x34, 0x05, 0x98, 0xf4, 0x98, 0xf2, 0x37, 0x56, 0x0a, 0x90, 0x0e, 0x79, 0x57, 0x27, 0x26, 0xad,
-	0x20, 0xa1, 0xe9, 0x2b, 0xe3, 0xbb, 0x4c, 0x8e, 0xd2, 0x98, 0x51, 0x6d, 0xca, 0x8b, 0x4f, 0x2c,
-	0xb1, 0x17, 0xee, 0xc0, 0xfc, 0xd0, 0x24, 0x44, 0x73, 0x90, 0xed, 0xd1, 0x43, 0xb9, 0x62, 0x60,
-	0xfe, 0x13, 0x5d, 0x81, 0xfc, 0x01, 0x31, 0x3d, 0x5a, 0xc9, 0x88, 0x32, 0xf9, 0xf1, 0xeb, 0x99,
-	0xdb, 0x5a, 0xed, 0x11, 0xcc, 0xac, 0x78, 0x6c, 0xdf, 0x76, 0x8c, 0x0f, 0xc5, 0x3c, 0x42, 0x1b,
-	0x90, 0x67, 0x76, 0x8f, 0x5a, 0x42, 0xbc, 0xbc, 0xfc, 0x46, 0x9a, 0x99, 0xe5, 0xd8, 0xbc, 0x4f,
-	0x0f, 0x7d, 0xbd, 0x8d, 0x12, 0xaf, 0xd9, 0x0e, 0x97, 0xc3, 0x52, 0xbc, 0xf6, 0x35, 0x28, 0x34,
-	0xbc, 0x4e, 0x87, 0x3a, 0xa7, 0x58, 0xc1, 0xea, 0x90, 0x63, 0x87, 0x03, 0x55, 0xbb, 0x60, 0x68,
-	0xe4, 0x76, 0x0e, 0x07, 0xf4, 0xd9, 0x51, 0x15, 0x24, 0x0e, 0xff, 0xc2, 0x82, 0xaf, 0xf6, 0x13,
-	0x0d, 0x2e, 0xcb, 0x42, 0x35, 0x7a, 0x56, 0x6d, 0xab, 0x63, 0x74, 0x11, 0x85, 0xbc, 0x43, 0xdb,
-	0x86, 0xab, 0xea, 0xbe, 0x36, 0xb6, 0xc9, 0x31, 0x47, 0x91, 0xa0, 0xb2, 0x69, 0xa2, 0x00, 0x4b,
-	0x74, 0xe4, 0x41, 0xe9, 0x03, 0xca, 0x5c, 0xe6, 0x50, 0xd2, 0x17, 0x75, 0x2e, 0x2f, 0xbf, 0x33,
-	0xb6, 0xaa, 0x7b, 0x94, 0xb5, 0x04, 0x92, 0x52, 0x37, 0x73, 0x7c, 0x54, 0x2d, 0x05, 0x85, 0x38,
-	0xd4, 0x54, 0xfb, 0x8f, 0x0c, 0x94, 0x82, 0xc5, 0x0b, 0xbd, 0x0e, 0x79, 0xe1, 0x2b, 0x94, 0x59,
-	0x83, 0xe1, 0x21, 0x5c, 0x0a, 0x96, 0x34, 0xf4, 0x06, 0x4c, 0xe9, 0x76, 0xbf, 0x4f, 0xac, 0x76,
-	0x25, 0x73, 0x23, 0x7b, 0xb3, 0xd4, 0x28, 0xf3, 0x59, 0xb1, 0x2a, 0x8b, 0xb0, 0x4f, 0x43, 0xd7,
-	0x21, 0x47, 0x9c, 0xae, 0x5b, 0xc9, 0x0a, 0x1e, 0xb1, 0x3a, 0xaf, 0x38, 0x5d, 0x17, 0x8b, 0x52,
-	0xf4, 0x25, 0xc8, 0x52, 0xeb, 0xa0, 0x92, 0x1b, 0x3d, 0xed, 0xd6, 0xad, 0x83, 0x87, 0xc4, 0x69,
-	0x94, 0x55, 0x1d, 0xb2, 0xeb, 0xd6, 0x01, 0xe6, 0x32, 0xe8, 0x3d, 0x98, 0x96, 0x33, 0x6f, 0x8b,
-	0x4f, 0x64, 0xb7, 0x92, 0x17, 0x18, 0xd5, 0xd1, 0x53, 0x57, 0xf0, 0x85, 0xab, 0x48, 0xa4, 0xd0,
-	0xc5, 0x31, 0x28, 0xf4, 0x1e, 0x94, 0xfc, 0x5d, 0x9e, 0xab, 0xd6, 0xe9, 0x54, 0x07, 0x8c, 0x15,
-	0x13, 0xa6, 0xdf, 0xf0, 0x0c, 0x87, 0xf6, 0xa9, 0xc5, 0xdc, 0xc6, 0xbc, 0x52, 0x50, 0xf2, 0xa9,
-	0x2e, 0x0e, 0xd1, 0x6a, 0xff, 0x9d, 0x81, 0xe1, 0x5d, 0x42, 0x5c, 0xa1, 0x76, 0x9e, 0x0a, 0xd1,
-	0x1e, 0xcc, 0x06, 0x7e, 0x7f, 0xdb, 0x36, 0x0d, 0xfd, 0x50, 0x4d, 0x85, 0xdb, 0x4a, 0x6c, 0xf6,
-	0x6e, 0x9c, 0xfc, 0xec, 0xa8, 0xfa, 0xda, 0xf0, 0x1e, 0xb9, 0x1e, 0x32, 0xe0, 0x24, 0x20, 0xd7,
-	0x91, 0x5c, 0x1e, 0xe5, 0x76, 0xf1, 0xf5, 0x11, 0x33, 0x7c, 0x8c, 0xb5, 0x71, 0xfc, 0x91, 0x52,
-	0xfb, 0x81, 0x06, 0xb9, 0xf5, 0x76, 0x97, 0x72, 0x6f, 0xd1, 0x71, 0xec, 0x7e, 0xd2, 0x5b, 0x6c,
-	0x38, 0x76, 0x1f, 0x0b, 0x0a, 0x5a, 0x80, 0x0c, 0xb3, 0x95, 0x81, 0x40, 0xd1, 0x33, 0x3b, 0x36,
-	0xce, 0x30, 0x1b, 0x7d, 0x08, 0xa0, 0xdb, 0x56, 0xdb, 0x90, 0x5b, 0x8b, 0xec, 0x84, 0x3b, 0xc8,
-	0x0d, 0xdb, 0x79, 0x42, 0x9c, 0xf6, 0x6a, 0x80, 0xd8, 0xb8, 0x74, 0x7c, 0x54, 0x85, 0xf0, 0x1b,
-	0x47, 0xb4, 0xd5, 0x6e, 0xc1, 0xfc, 0x90, 0x00, 0xaa, 0x42, 0xbe, 0x47, 0x0f, 0xef, 0x72, 0x77,
-	0xca, 0xe7, 0x96, 0x70, 0x26, 0xf7, 0x79, 0x01, 0x96, 0xe5, 0xb5, 0xff, 0xd5, 0xa0, 0xb8, 0xe1,
-	0x59, 0xba, 0x70, 0xbe, 0xcf, 0x77, 0x95, 0xfe, 0x54, 0xcd, 0xa4, 0x4e, 0x55, 0x0f, 0x0a, 0xbd,
-	0x27, 0xc1, 0x54, 0x2e, 0x2f, 0x6f, 0x8d, 0xdf, 0x74, 0x55, 0xa5, 0xfa, 0x7d, 0x81, 0x27, 0x77,
-	0x77, 0x97, 0x54, 0x85, 0x0a, 0xf7, 0x1f, 0x09, 0xa5, 0x4a, 0xd9, 0xc2, 0x97, 0xa0, 0x1c, 0x61,
-	0x3b, 0xd3, 0xfa, 0xf3, 0x37, 0x1a, 0xcc, 0x6e, 0xca, 0x28, 0xc8, 0x76, 0x64, 0xcc, 0x81, 0xae,
-	0x41, 0xd6, 0x19, 0x78, 0x42, 0x3e, 0x2b, 0xb7, 0xcf, 0x78, 0x7b, 0x17, 0xf3, 0x32, 0xf4, 0x5b,
-	0x50, 0x6c, 0x7b, 0x72, 0xc7, 0xa7, 0x3c, 0x6f, 0x3d, 0x32, 0xcc, 0x82, 0x58, 0x2b, 0x6c, 0x19,
-	0xdf, 0x0b, 0xf3, 0x81, 0xb7, 0xa6, 0xa4, 0xe4, 0x66, 0xc5, 0xff, 0xc2, 0x01, 0x1a, 0x77, 0x95,
-	0x7d, 0xb7, 0xdb, 0x32, 0x3e, 0x94, 0x61, 0x54, 0x5e, 0xba, 0xca, 0x2d, 0x59, 0x84, 0x7d, 0x5a,
-	0xed, 0xdb, 0x19, 0xb8, 0xba, 0x49, 0xd9, 0x1a, 0xa1, 0x7d, 0xdb, 0x5a, 0xa3, 0x03, 0xd3, 0x3e,
-	0xe4, 0x33, 0x1c, 0xd3, 0x6f, 0xa0, 0xaf, 0x02, 0x18, 0xee, 0x5e, 0xeb, 0x40, 0xe7, 0x2b, 0x95,
-	0xea, 0xc2, 0x1b, 0xca, 0x62, 0x70, 0xb7, 0xd5, 0x50, 0x94, 0x67, 0xb1, 0x2f, 0x1c, 0x91, 0x09,
-	0x7d, 0x7a, 0xe6, 0x04, 0x9f, 0xde, 0x02, 0x18, 0x84, 0x7e, 0x22, 0x2b, 0x38, 0x7f, 0xd5, 0x57,
-	0x73, 0x16, 0x17, 0x11, 0x81, 0x99, 0x64, 0xe6, 0xfe, 0x43, 0x16, 0x16, 0x36, 0x29, 0x0b, 0x96,
-	0x2c, 0xb5, 0x24, 0xb7, 0x06, 0x54, 0xe7, 0x56, 0xf9, 0x48, 0x83, 0x82, 0x49, 0xf6, 0xa8, 0xe9,
-	0x8a, 0x29, 0x50, 0x5e, 0x7e, 0x3c, 0xf6, 0x98, 0x1c, 0xad, 0xa5, 0xde, 0x14, 0x1a, 0x12, 0xa3,
-	0x54, 0x16, 0x62, 0xa5, 0x1e, 0x7d, 0x01, 0xca, 0xba, 0xe9, 0xb9, 0x8c, 0x3a, 0xdb, 0xb6, 0xc3,
-	0x84, 0x8d, 0xf3, 0x61, 0x5c, 0xb1, 0x1a, 0x92, 0x70, 0x94, 0x0f, 0x2d, 0x03, 0xe8, 0xa6, 0x41,
-	0x2d, 0x26, 0xa4, 0xe4, 0xd8, 0x40, 0xbe, 0xbd, 0x57, 0x03, 0x0a, 0x8e, 0x70, 0x71, 0x55, 0x7d,
-	0xdb, 0x32, 0x98, 0x2d, 0x55, 0xe5, 0xe2, 0xaa, 0xb6, 0x42, 0x12, 0x8e, 0xf2, 0x09, 0x31, 0xca,
-	0x1c, 0x43, 0x77, 0x85, 0x58, 0x3e, 0x21, 0x16, 0x92, 0x70, 0x94, 0x8f, 0x4f, 0xbf, 0x48, 0xfb,
-	0xcf, 0x34, 0xfd, 0xfe, 0xb1, 0x08, 0x8b, 0x31, 0xb3, 0x32, 0xc2, 0x68, 0xc7, 0x33, 0x5b, 0x94,
-	0xf9, 0x1d, 0xf8, 0x05, 0x28, 0xab, 0xfd, 0xf8, 0x83, 0xd0, 0x35, 0x05, 0x95, 0x6a, 0x85, 0x24,
-	0x1c, 0xe5, 0x43, 0xdf, 0x0a, 0xfb, 0x3d, 0x23, 0xfa, 0x5d, 0x3f, 0x9f, 0x7e, 0x1f, 0xaa, 0xe0,
-	0xa9, 0xfa, 0x7e, 0x09, 0x4a, 0x16, 0x61, 0xae, 0x98, 0x48, 0x6a, 0xce, 0x04, 0x4b, 0xf2, 0x03,
-	0x9f, 0x80, 0x43, 0x1e, 0xb4, 0x0d, 0x57, 0x94, 0x89, 0xd7, 0x9f, 0x0e, 0x6c, 0x87, 0x51, 0x47,
-	0xca, 0xe6, 0x84, 0xec, 0x75, 0x25, 0x7b, 0x65, 0x2b, 0x85, 0x07, 0xa7, 0x4a, 0xa2, 0x2d, 0xb8,
-	0xac, 0x8b, 0x2d, 0x1e, 0xa6, 0xa6, 0x4d, 0xda, 0x3e, 0x60, 0x5e, 0x00, 0xfe, 0xbc, 0x02, 0xbc,
-	0xbc, 0x3a, 0xcc, 0x82, 0xd3, 0xe4, 0x92, 0xa3, 0xb9, 0x30, 0xd6, 0x68, 0x9e, 0x1a, 0x67, 0x34,
-	0x17, 0xc7, 0x1b, 0xcd, 0xa5, 0xd3, 0x8d, 0x66, 0x6e, 0x79, 0x3e, 0x8e, 0xa8, 0xc3, 0xe3, 0x12,
-	0x19, 0x69, 0x88, 0x81, 0x07, 0x71, 0xcb, 0xb7, 0x52, 0x78, 0x70, 0xaa, 0x24, 0xda, 0x83, 0x05,
-	0x59, 0xbe, 0x6e, 0xe9, 0xce, 0xe1, 0x80, 0xbb, 0xfb, 0x08, 0x6e, 0x59, 0xe0, 0xd6, 0x14, 0xee,
-	0x42, 0x6b, 0x24, 0x27, 0x3e, 0x01, 0x05, 0xfd, 0x06, 0xcc, 0xc8, 0x5e, 0xda, 0x22, 0x83, 0x48,
-	0x88, 0xfe, 0xaa, 0x82, 0x9d, 0x59, 0x8d, 0x12, 0x71, 0x9c, 0x17, 0xad, 0xc0, 0xec, 0xe0, 0x40,
-	0xe7, 0x3f, 0xef, 0x76, 0x1e, 0x50, 0xda, 0xa6, 0x6d, 0x11, 0xa1, 0x97, 0x1a, 0x3f, 0xe7, 0xef,
-	0xff, 0xb6, 0xe3, 0x64, 0x9c, 0xe4, 0x47, 0xb7, 0x61, 0xda, 0x65, 0xc4, 0x61, 0x6a, 0x6f, 0x2f,
-	0xe2, 0xf6, 0x52, 0xb8, 0x91, 0x6e, 0x45, 0x68, 0x38, 0xc6, 0x39, 0x89, 0xf7, 0x78, 0x26, 0x17,
-	0x43, 0x11, 0x1c, 0x25, 0xdc, 0xfe, 0x1f, 0x24, 0xdd, 0xfe, 0xfb, 0x93, 0x4c, 0xff, 0x14, 0x0d,
-	0xa7, 0x9a, 0xf6, 0xf7, 0x00, 0x39, 0x2a, 0x94, 0x93, 0xbb, 0xf9, 0x88, 0xe7, 0x0f, 0x32, 0x10,
-	0x78, 0x88, 0x03, 0xa7, 0x48, 0xa1, 0x16, 0xbc, 0xea, 0x52, 0x8b, 0x19, 0x16, 0x35, 0xe3, 0x70,
-	0x72, 0x49, 0x78, 0x4d, 0xc1, 0xbd, 0xda, 0x4a, 0x63, 0xc2, 0xe9, 0xb2, 0x93, 0x18, 0xff, 0x87,
-	0x25, 0xb1, 0xee, 0x4a, 0xd3, 0x9c, 0x9b, 0xdb, 0xfe, 0x28, 0xe9, 0xb6, 0x1f, 0x4f, 0xde, 0x6f,
-	0xe3, 0xb9, 0xec, 0x65, 0x00, 0xd1, 0x0b, 0x51, 0x9f, 0x1d, 0x78, 0x2a, 0x1c, 0x50, 0x70, 0x84,
-	0x8b, 0xcf, 0x42, 0xdf, 0xce, 0x51, 0x77, 0x1d, 0xcc, 0xc2, 0x56, 0x94, 0x88, 0xe3, 0xbc, 0x23,
-	0x5d, 0x7e, 0x7e, 0x6c, 0x97, 0x7f, 0x0f, 0x90, 0x61, 0x19, 0x2c, 0xe8, 0x72, 0x89, 0x57, 0x88,
-	0x27, 0xc0, 0xee, 0x0e, 0x71, 0xe0, 0x14, 0xa9, 0x11, 0x43, 0x79, 0xea, 0x7c, 0x87, 0x72, 0x71,
-	0xfc, 0xa1, 0x8c, 0x1e, 0xc3, 0x35, 0xa1, 0x4a, 0xd9, 0x27, 0x0e, 0x2c, 0x9d, 0xff, 0x2f, 0x28,
-	0xe0, 0x6b, 0x78, 0x14, 0x23, 0x1e, 0x8d, 0xc1, 0xfb, 0x47, 0x77, 0x68, 0x9b, 0x2b, 0x27, 0xe6,
-	0xe8, 0x85, 0x61, 0x35, 0x85, 0x07, 0xa7, 0x4a, 0xf2, 0x21, 0xc6, 0xf8, 0x30, 0x24, 0x7b, 0x26,
-	0x6d, 0x8b, 0x85, 0xa0, 0x18, 0x0e, 0xb1, 0x9d, 0x66, 0x4b, 0x51, 0x70, 0x84, 0x2b, 0xcd, 0x57,
-	0x4f, 0x9f, 0xd1, 0x57, 0x6f, 0x8a, 0xd3, 0x8e, 0x4e, 0x6c, 0x49, 0x50, 0x0e, 0x3f, 0x48, 0xe9,
-	0xae, 0x26, 0x19, 0xf0, 0xb0, 0x8c, 0x58, 0x2a, 0x75, 0xc7, 0x18, 0x30, 0x37, 0x8e, 0x75, 0x29,
-	0xb1, 0x54, 0xa6, 0xf0, 0xe0, 0x54, 0x49, 0xbe, 0x49, 0xd9, 0xa7, 0xc4, 0x64, 0xfb, 0x71, 0xc0,
-	0xd9, 0xf8, 0x26, 0xe5, 0x9d, 0x61, 0x16, 0x9c, 0x26, 0x37, 0x89, 0x7b, 0xfb, 0xe3, 0x0c, 0x5c,
-	0xde, 0xa4, 0xea, 0xa4, 0x61, 0xdb, 0x6e, 0xfb, 0x7e, 0xed, 0x67, 0x34, 0xca, 0xfa, 0x33, 0x0d,
-	0xe0, 0x9d, 0x9d, 0x9d, 0x6d, 0x15, 0x22, 0xb7, 0x21, 0x47, 0x3c, 0xb6, 0xaf, 0xf2, 0x50, 0x1b,
-	0xe3, 0x1f, 0xe8, 0x44, 0x73, 0xbf, 0x2a, 0x9d, 0xe0, 0xb1, 0x7d, 0x2c, 0xd0, 0xd1, 0x2f, 0xc2,
-	0x94, 0x5a, 0x1b, 0x84, 0xad, 0x8a, 0x61, 0x62, 0x5d, 0xad, 0x1f, 0xd8, 0xa7, 0xd7, 0x7e, 0x9c,
-	0x81, 0xab, 0x77, 0x2d, 0x46, 0x9d, 0x16, 0xa3, 0x83, 0x58, 0x6e, 0x16, 0xfd, 0x4e, 0xe4, 0xc8,
-	0x4b, 0xd6, 0xf7, 0x57, 0x4e, 0x17, 0xb3, 0xcb, 0x63, 0x93, 0x2d, 0xca, 0x48, 0x38, 0x2b, 0xc3,
-	0xb2, 0xc8, 0x39, 0x97, 0x07, 0x39, 0x77, 0x40, 0x75, 0x95, 0x11, 0x68, 0x8d, 0x6d, 0x8d, 0xf4,
-	0x06, 0xf0, 0x91, 0x17, 0xe6, 0x62, 0xc4, 0x38, 0x14, 0xea, 0xd0, 0x37, 0xa1, 0xe0, 0x32, 0xc2,
-	0x3c, 0x3f, 0xd1, 0xb4, 0x7b, 0xde, 0x8a, 0x05, 0x78, 0xb8, 0x40, 0xca, 0x6f, 0xac, 0x94, 0xd6,
-	0x7e, 0xac, 0xc1, 0x42, 0xba, 0x60, 0xd3, 0x70, 0x19, 0xfa, 0xfa, 0x90, 0xd9, 0x4f, 0x99, 0x2a,
-	0xe1, 0xd2, 0xc2, 0xe8, 0x73, 0x4a, 0x71, 0xd1, 0x2f, 0x89, 0x98, 0x9c, 0x41, 0xde, 0x60, 0xb4,
-	0xef, 0xef, 0x12, 0xde, 0x3d, 0xe7, 0xa6, 0x47, 0x66, 0x25, 0xd7, 0x82, 0xa5, 0xb2, 0xda, 0x47,
-	0x99, 0x51, 0x4d, 0xe6, 0xdd, 0x82, 0x7a, 0xf1, 0xfc, 0xff, 0xbd, 0xc9, 0xf2, 0xff, 0x0d, 0x2f,
-	0x52, 0x9f, 0xe1, 0x53, 0x80, 0xdf, 0x1d, 0x3e, 0x05, 0x78, 0x77, 0xf2, 0x53, 0x80, 0x84, 0x15,
-	0x46, 0x1e, 0x06, 0xfc, 0x30, 0x03, 0xd7, 0x4f, 0x1a, 0x35, 0xa8, 0x1b, 0x0c, 0x4e, 0x6d, 0xd2,
-	0x5b, 0x01, 0x27, 0x0e, 0x43, 0xb4, 0x0c, 0xf9, 0xc1, 0x3e, 0x71, 0x7d, 0x77, 0xea, 0xaf, 0x3a,
-	0xf9, 0x6d, 0x5e, 0xf8, 0xec, 0xa8, 0x5a, 0x96, 0x6e, 0x58, 0x7c, 0x62, 0xc9, 0xca, 0x1d, 0x4b,
-	0x9f, 0xba, 0x6e, 0xb8, 0xb1, 0x0b, 0x1c, 0xcb, 0x96, 0x2c, 0xc6, 0x3e, 0x1d, 0x31, 0x28, 0xc8,
-	0x60, 0x49, 0x5d, 0x3d, 0x68, 0x8e, 0xdd, 0x8e, 0x94, 0x13, 0xa3, 0xb0, 0x51, 0x2a, 0xee, 0x56,
-	0xba, 0x6a, 0x7f, 0x7b, 0x09, 0xae, 0xa6, 0xf7, 0x09, 0xaf, 0xfb, 0x01, 0x75, 0x5c, 0xc3, 0xb6,
-	0xd4, 0xea, 0x13, 0x9e, 0x36, 0xca, 0x62, 0xec, 0xd3, 0xd1, 0x4d, 0x28, 0x3a, 0x74, 0x60, 0x1a,
-	0x3a, 0x71, 0x55, 0xd0, 0x21, 0xb2, 0x8f, 0x58, 0x95, 0xe1, 0x80, 0x3a, 0xe2, 0x06, 0x44, 0xf6,
-	0xa7, 0x78, 0x03, 0xe2, 0x2f, 0x35, 0xbe, 0x9f, 0x93, 0x19, 0x87, 0x21, 0x01, 0xd5, 0x17, 0xe7,
-	0x59, 0xb3, 0xd7, 0xe4, 0xbe, 0x70, 0x84, 0x42, 0x3c, 0xba, 0x2e, 0xe8, 0x7b, 0x1a, 0x54, 0xfa,
-	0x89, 0x0d, 0xe3, 0x05, 0x5e, 0x22, 0xb9, 0x7e, 0x7c, 0x54, 0xad, 0x6c, 0x8d, 0xd0, 0x87, 0x47,
-	0xd6, 0x04, 0xfd, 0x1e, 0x94, 0x07, 0x7c, 0x5c, 0xb8, 0x8c, 0x5a, 0x3a, 0x55, 0x47, 0x56, 0xe3,
-	0x8f, 0xe6, 0xed, 0x10, 0xab, 0xc5, 0x1c, 0xc2, 0x68, 0xf7, 0xb0, 0x31, 0xcb, 0x43, 0xbb, 0x08,
-	0x01, 0x47, 0x35, 0xc6, 0xae, 0x9e, 0x6c, 0x5d, 0xf4, 0xd5, 0x93, 0xef, 0xa6, 0x5f, 0x3d, 0x21,
-	0xe7, 0xec, 0x21, 0x5f, 0x5e, 0x41, 0x79, 0x79, 0x05, 0xe5, 0x45, 0x5d, 0x41, 0xb9, 0x09, 0x45,
-	0x97, 0x32, 0x66, 0x58, 0x5d, 0xb7, 0x32, 0x27, 0x0f, 0xe8, 0xb8, 0xd6, 0x96, 0x2a, 0xc3, 0x01,
-	0x15, 0xfd, 0x32, 0x94, 0x44, 0x8a, 0x6d, 0xc5, 0xe9, 0xba, 0x95, 0x79, 0x71, 0x52, 0x27, 0x56,
-	0xf2, 0x96, 0x5f, 0x88, 0x43, 0x3a, 0xba, 0x05, 0xd3, 0x7b, 0x62, 0x48, 0xcb, 0x25, 0x48, 0x5c,
-	0x17, 0x29, 0x35, 0xe6, 0xf8, 0x08, 0x6e, 0x44, 0xca, 0x71, 0x8c, 0x8b, 0x87, 0xae, 0x34, 0xc8,
-	0x43, 0x56, 0x2e, 0xc7, 0x43, 0xd7, 0x30, 0x43, 0x89, 0x23, 0x5c, 0xe8, 0x35, 0xc8, 0x32, 0xd3,
-	0xad, 0x5c, 0x11, 0xcc, 0x41, 0x88, 0xb1, 0xd3, 0x6c, 0x61, 0x5e, 0x3e, 0xf9, 0x5d, 0x92, 0xff,
-	0xd3, 0x60, 0x36, 0x71, 0x9d, 0x81, 0xeb, 0xf4, 0x1c, 0x53, 0xad, 0x94, 0x81, 0xce, 0x5d, 0xdc,
-	0xc4, 0xbc, 0x1c, 0x3d, 0x56, 0x71, 0x4c, 0x66, 0x42, 0x7f, 0xf4, 0x60, 0x65, 0xa7, 0xc5, 0x03,
-	0x97, 0xa1, 0x10, 0xe6, 0x76, 0xc2, 0xba, 0xd9, 0x78, 0x5e, 0xf4, 0x64, 0x0b, 0x47, 0x92, 0x03,
-	0xb9, 0xd3, 0x24, 0x07, 0x6a, 0x9f, 0x6a, 0x50, 0xba, 0x4f, 0x3a, 0x3d, 0xd2, 0x32, 0xac, 0x1e,
-	0x7a, 0x03, 0xa6, 0xf6, 0x1c, 0xbb, 0x47, 0x1d, 0x57, 0x9d, 0xfe, 0x8a, 0x23, 0xc5, 0x86, 0x2c,
-	0xc2, 0x3e, 0x8d, 0xc7, 0xa3, 0xcc, 0x1e, 0x18, 0x7a, 0x32, 0x1e, 0xdd, 0xe1, 0x85, 0x58, 0xd2,
-	0xd0, 0x23, 0xd9, 0x77, 0xd9, 0x09, 0x2f, 0x24, 0xee, 0x34, 0x5b, 0xf2, 0x44, 0xd5, 0xef, 0x75,
-	0xf4, 0x66, 0x6c, 0x7f, 0x55, 0x1a, 0xb9, 0x23, 0xfa, 0x6e, 0x06, 0xca, 0xb2, 0x69, 0x32, 0x02,
-	0x3d, 0xcf, 0xc6, 0xdd, 0x11, 0xc9, 0x73, 0xd7, 0xeb, 0x53, 0x67, 0xd3, 0xb1, 0xbd, 0x81, 0xea,
-	0xa5, 0x68, 0x32, 0x24, 0x24, 0x06, 0x09, 0xf4, 0xb0, 0xc8, 0xb7, 0x4e, 0xee, 0x02, 0xad, 0x93,
-	0x3f, 0xd1, 0x3a, 0x7f, 0xa7, 0x41, 0xa9, 0x69, 0x74, 0xa8, 0x7e, 0xa8, 0x9b, 0x14, 0x7d, 0x1d,
-	0x2a, 0x6d, 0x6a, 0x52, 0x46, 0x37, 0x1d, 0xa2, 0xd3, 0x6d, 0xea, 0x18, 0xc2, 0xd5, 0xdb, 0x56,
-	0x5b, 0xee, 0xc6, 0xf3, 0x41, 0xc6, 0xa2, 0xb2, 0x36, 0x82, 0x0f, 0x8f, 0x44, 0x40, 0x77, 0x61,
-	0xba, 0x4d, 0x5d, 0xc3, 0xa1, 0xed, 0xed, 0xc8, 0xbe, 0xfb, 0x0d, 0x7f, 0x48, 0xaf, 0x45, 0x68,
-	0xcf, 0x8e, 0xaa, 0x33, 0xdb, 0xc6, 0x80, 0x9a, 0x86, 0x45, 0xe5, 0x06, 0x3c, 0x26, 0x5a, 0xcb,
-	0x43, 0xb6, 0x69, 0x77, 0x6b, 0x7f, 0x98, 0x85, 0x60, 0x0d, 0x47, 0x7f, 0xa4, 0x41, 0x99, 0x58,
-	0x96, 0xcd, 0xd4, 0xe2, 0x28, 0xd3, 0xf7, 0x78, 0xe2, 0xad, 0x42, 0x7d, 0x25, 0x04, 0x95, 0x2b,
-	0x75, 0xb0, 0xa0, 0x46, 0x28, 0x38, 0xaa, 0x1b, 0x79, 0x89, 0x64, 0xf4, 0xd6, 0xe4, 0xb5, 0x38,
-	0x45, 0xea, 0x79, 0xe1, 0x2b, 0x30, 0x97, 0xac, 0xec, 0x59, 0x1c, 0xe1, 0x24, 0x69, 0xaf, 0xbf,
-	0xd0, 0xa0, 0xe8, 0x3b, 0x33, 0xb4, 0x0a, 0x39, 0xcf, 0xa5, 0xce, 0xd9, 0xae, 0xe2, 0x09, 0x0f,
-	0xb8, 0xeb, 0x52, 0x07, 0x0b, 0x61, 0xf4, 0x2e, 0x14, 0x07, 0xc4, 0x75, 0x9f, 0xd8, 0x4e, 0x5b,
-	0xb9, 0xd9, 0x53, 0x02, 0xc9, 0xb5, 0x59, 0x89, 0xe2, 0x00, 0xa4, 0xf6, 0xfd, 0x19, 0x28, 0x3f,
-	0x20, 0xcc, 0x38, 0xa0, 0x22, 0x1e, 0xbe, 0x98, 0x80, 0xe8, 0xcf, 0x35, 0xb8, 0x1a, 0xcf, 0x5c,
-	0x5f, 0x60, 0x54, 0xb4, 0x70, 0x7c, 0x54, 0xbd, 0x8a, 0x53, 0xb5, 0xe1, 0x11, 0xb5, 0x10, 0xf1,
-	0xd1, 0x50, 0x22, 0xfc, 0xa2, 0xe3, 0xa3, 0xd6, 0x28, 0x85, 0x78, 0x74, 0x5d, 0x5e, 0xc6, 0x47,
-	0x63, 0xc4, 0x47, 0x17, 0x7e, 0x35, 0xff, 0x3b, 0xe9, 0xf1, 0xd1, 0xc3, 0xf1, 0x77, 0x40, 0xe1,
-	0x8c, 0x7c, 0x19, 0x14, 0xbd, 0x0c, 0x8a, 0x5e, 0x54, 0x50, 0x34, 0x48, 0x04, 0x45, 0x93, 0x1c,
-	0x46, 0xa8, 0x53, 0x7e, 0x89, 0x36, 0x2a, 0xb8, 0x9a, 0x3c, 0x4c, 0xf9, 0xd3, 0x0c, 0x5c, 0x4e,
-	0xf1, 0x0e, 0xe8, 0xab, 0x30, 0xe7, 0x32, 0xdb, 0x21, 0x5d, 0x1a, 0x76, 0xa8, 0x5c, 0xd0, 0xae,
-	0xf0, 0x31, 0xd1, 0x4a, 0xd0, 0xf0, 0x10, 0x37, 0x7a, 0x0c, 0x40, 0x74, 0x9d, 0xba, 0xee, 0x96,
-	0xdd, 0xf6, 0xf7, 0x65, 0x77, 0x78, 0xb8, 0xb0, 0x12, 0x94, 0x3e, 0x3b, 0xaa, 0x7e, 0x3e, 0xed,
-	0xc0, 0xc8, 0xaf, 0x0f, 0x93, 0x57, 0x9e, 0x43, 0x01, 0x1c, 0x81, 0x44, 0xbf, 0x0d, 0x20, 0x2f,
-	0x41, 0x07, 0xf7, 0x14, 0x9f, 0x93, 0xd5, 0xaf, 0xfb, 0x97, 0x8c, 0xeb, 0xbf, 0xe9, 0x11, 0x8b,
-	0xf1, 0x51, 0x21, 0xae, 0xb0, 0x3e, 0x0c, 0x50, 0x70, 0x04, 0xb1, 0xf6, 0x4f, 0x19, 0x28, 0xfa,
-	0xfb, 0xc5, 0x17, 0x70, 0x6e, 0xd3, 0x8d, 0x9d, 0xdb, 0x8c, 0xff, 0x16, 0xc3, 0xaf, 0xf2, 0xc8,
-	0x93, 0x1a, 0x3b, 0x71, 0x52, 0xb3, 0x39, 0xb9, 0xaa, 0x93, 0xcf, 0x66, 0x9e, 0x69, 0x70, 0xc9,
-	0x67, 0x95, 0xef, 0x42, 0xd0, 0x17, 0x61, 0xc6, 0xa1, 0xa4, 0xdd, 0x20, 0x4c, 0xdf, 0x17, 0xdd,
-	0xc7, 0x6d, 0x9a, 0x6b, 0xcc, 0x1f, 0x1f, 0x55, 0x67, 0x70, 0x94, 0x80, 0xe3, 0x7c, 0xa8, 0x0e,
-	0xe0, 0xb5, 0x3b, 0x8f, 0x6c, 0x47, 0x04, 0x5b, 0xdc, 0x56, 0x33, 0xb2, 0x13, 0x77, 0xd7, 0x36,
-	0x54, 0x29, 0x8e, 0x70, 0xa0, 0xb7, 0x61, 0x56, 0x06, 0xb2, 0x5b, 0xe4, 0x69, 0x93, 0x5a, 0x5d,
-	0xb6, 0x2f, 0x5a, 0x9d, 0x93, 0x8e, 0xb4, 0x11, 0x27, 0xe1, 0x24, 0x2f, 0x9f, 0x06, 0xb2, 0x68,
-	0xd7, 0x25, 0x5d, 0x59, 0x79, 0xb1, 0x9d, 0x99, 0x91, 0xd3, 0xa0, 0x91, 0xa0, 0xe1, 0x21, 0xee,
-	0xda, 0xbf, 0x68, 0x30, 0x1d, 0x36, 0xfe, 0xc2, 0x8f, 0xa2, 0x3a, 0xf1, 0xa3, 0xa8, 0x95, 0x89,
-	0xfb, 0x76, 0xc4, 0xe1, 0xd3, 0xf7, 0xf2, 0x61, 0xb3, 0xc4, 0x71, 0xd3, 0x1e, 0x2c, 0x18, 0xa9,
-	0x47, 0x30, 0x11, 0xd7, 0x11, 0xdc, 0x2b, 0xbb, 0x3b, 0x92, 0x13, 0x9f, 0x80, 0x82, 0x3c, 0x28,
-	0x1e, 0x50, 0x87, 0x19, 0x3a, 0xf5, 0xdb, 0xb7, 0x79, 0x4e, 0xaf, 0xf7, 0x42, 0x9b, 0x3e, 0x54,
-	0x0a, 0x70, 0xa0, 0x0a, 0xed, 0x41, 0x9e, 0xb6, 0xbb, 0xd4, 0xbf, 0x47, 0x3e, 0xfe, 0x7b, 0xcf,
-	0xf5, 0x76, 0x37, 0x62, 0x4f, 0xfe, 0xe5, 0x62, 0x09, 0x8d, 0x5c, 0x28, 0x99, 0x7e, 0xc8, 0xac,
-	0x36, 0xcc, 0x8d, 0xb1, 0xf5, 0x04, 0xc1, 0x77, 0x78, 0xaf, 0x33, 0x28, 0xc2, 0xa1, 0x1e, 0xd4,
-	0x0b, 0x1e, 0x80, 0xe5, 0xcf, 0xc9, 0x13, 0x9c, 0xf0, 0x04, 0xcc, 0x85, 0xd2, 0x13, 0xc2, 0xa8,
-	0xd3, 0x27, 0x4e, 0x4f, 0x6d, 0x68, 0xc7, 0x6f, 0xe1, 0x23, 0x1f, 0x29, 0x6c, 0x61, 0x50, 0x84,
-	0x43, 0x3d, 0xb5, 0xef, 0x67, 0x42, 0xd7, 0xf3, 0xa2, 0xcf, 0x02, 0x6f, 0xc5, 0xcf, 0x02, 0x17,
-	0x93, 0x67, 0x81, 0x89, 0x64, 0xc4, 0xd9, 0x4f, 0x03, 0x09, 0x94, 0x4d, 0xe2, 0xb2, 0xdd, 0x41,
-	0x9b, 0x30, 0x95, 0x95, 0x2b, 0x2f, 0xff, 0xd2, 0xe9, 0x9c, 0xc9, 0x8e, 0xd1, 0xa7, 0xe1, 0x7e,
-	0xb5, 0x19, 0xc2, 0xe0, 0x28, 0x66, 0xed, 0x3f, 0x35, 0x98, 0x1f, 0x3a, 0xff, 0x45, 0xfb, 0x50,
-	0xb0, 0xc4, 0x0e, 0x7b, 0xe2, 0xb7, 0x65, 0x91, 0x8d, 0xba, 0x1c, 0x34, 0xaa, 0x40, 0xe1, 0x23,
-	0x0b, 0x8a, 0xf4, 0x29, 0xa3, 0x8e, 0x45, 0x4c, 0xb5, 0x30, 0x9e, 0xcf, 0x3b, 0x36, 0xb1, 0x9f,
-	0x5a, 0x57, 0xc8, 0x38, 0xd0, 0x51, 0xfb, 0x49, 0x06, 0xca, 0x11, 0xbe, 0xe7, 0x65, 0x6c, 0xc5,
-	0xbd, 0x42, 0x19, 0x6a, 0xee, 0x3a, 0xa6, 0xea, 0xe8, 0xc8, 0xbd, 0x42, 0x45, 0xc2, 0x4d, 0x1c,
-	0xe5, 0x43, 0xcb, 0x00, 0x7d, 0xe2, 0x32, 0xea, 0x08, 0xdf, 0x98, 0xb8, 0xcd, 0xb7, 0x15, 0x50,
-	0x70, 0x84, 0x0b, 0xdd, 0x50, 0xe9, 0x8f, 0x5c, 0xfc, 0x35, 0xcc, 0x88, 0xdc, 0x46, 0xfe, 0x1c,
-	0x72, 0x1b, 0xa8, 0x0b, 0x73, 0x7e, 0xad, 0x7d, 0xaa, 0x9a, 0xb8, 0xa7, 0x04, 0x96, 0x5b, 0xc5,
-	0x04, 0x04, 0x1e, 0x02, 0xad, 0xfd, 0xbd, 0x06, 0x33, 0xb1, 0xfd, 0x2e, 0x7a, 0x3d, 0x7a, 0x79,
-	0x21, 0x92, 0x29, 0x8d, 0x5d, 0x3a, 0x78, 0x13, 0x0a, 0xd2, 0x40, 0xca, 0xf0, 0xc1, 0x44, 0x94,
-	0x26, 0xc4, 0x8a, 0xca, 0xa7, 0x94, 0x4a, 0xa5, 0x24, 0xa7, 0x94, 0xca, 0xb5, 0x60, 0x9f, 0x8e,
-	0x3e, 0xc7, 0x77, 0xf0, 0xb2, 0x76, 0xca, 0xd2, 0xc1, 0xc2, 0xe0, 0xb7, 0x03, 0x07, 0x1c, 0xb5,
-	0xb7, 0x41, 0x3e, 0x40, 0x45, 0xd7, 0x20, 0xdb, 0x37, 0x2c, 0x95, 0xce, 0x14, 0x49, 0xd3, 0x2d,
-	0xc3, 0xc2, 0xbc, 0x4c, 0x90, 0xc8, 0x53, 0x95, 0xe0, 0x91, 0x24, 0xf2, 0x14, 0xf3, 0xb2, 0xda,
-	0x5f, 0x67, 0x40, 0x3c, 0xfc, 0x47, 0x8f, 0x20, 0x6b, 0xda, 0x5d, 0x35, 0x99, 0xc6, 0xcf, 0xd8,
-	0x36, 0xed, 0xae, 0xd4, 0xd0, 0xb4, 0xbb, 0x98, 0x23, 0x22, 0x1d, 0xf2, 0x3d, 0xd2, 0xe9, 0x11,
-	0x35, 0x77, 0xc6, 0xf7, 0xb7, 0x41, 0x1e, 0x5f, 0x3d, 0xda, 0xe2, 0x9f, 0x58, 0x62, 0x23, 0x1d,
-	0x0a, 0x5e, 0x5b, 0xfc, 0x1f, 0xc2, 0xa4, 0x7f, 0xb9, 0xb0, 0xbb, 0x26, 0x54, 0x08, 0x47, 0x20,
-	0x7f, 0x63, 0x05, 0x5d, 0xfb, 0xe7, 0x0c, 0xa8, 0x7f, 0x61, 0x40, 0x1e, 0x94, 0xba, 0xfe, 0x23,
-	0x29, 0x65, 0xb3, 0x77, 0x26, 0xb8, 0x97, 0x1b, 0x7b, 0x6e, 0x25, 0x8f, 0xa6, 0x82, 0x42, 0x1c,
-	0x6a, 0x42, 0x34, 0x6e, 0xcb, 0xb5, 0x09, 0x6d, 0x29, 0xd5, 0x0d, 0x5b, 0x93, 0x40, 0x6e, 0x9f,
-	0xb1, 0x81, 0xb2, 0xe5, 0xea, 0xd8, 0x5a, 0xc2, 0xfb, 0x71, 0x32, 0x09, 0xca, 0xbf, 0xb1, 0x80,
-	0xae, 0xf5, 0x41, 0x2d, 0x55, 0x48, 0x8f, 0xbd, 0x10, 0x94, 0xc9, 0xed, 0xa5, 0xd3, 0x2d, 0x20,
-	0xc1, 0xb3, 0xbe, 0xc8, 0x8b, 0x89, 0xf4, 0xa7, 0x80, 0xff, 0x96, 0x81, 0xec, 0x4e, 0xb3, 0x25,
-	0x2f, 0x00, 0x8b, 0x4c, 0x05, 0x6d, 0xf5, 0x8c, 0xc1, 0x43, 0xea, 0x18, 0x1d, 0x19, 0xa5, 0x16,
-	0xa3, 0x17, 0x80, 0x93, 0x1c, 0x38, 0x45, 0x0a, 0xbd, 0x0f, 0xd3, 0x3a, 0x59, 0xa5, 0x0e, 0x93,
-	0x4e, 0xe7, 0x6c, 0xb9, 0x5c, 0x71, 0x9c, 0xb8, 0xba, 0x12, 0x8a, 0xe3, 0x18, 0x18, 0xda, 0x05,
-	0xd0, 0x43, 0xe8, 0xec, 0x59, 0xa0, 0xe5, 0x93, 0xc8, 0x10, 0x38, 0x02, 0x84, 0x30, 0x94, 0x7a,
-	0x9c, 0x55, 0xa0, 0xe6, 0xce, 0x82, 0x2a, 0x06, 0xe5, 0x7d, 0x5f, 0x16, 0x87, 0x30, 0xb5, 0xbf,
-	0xd2, 0xa0, 0xb8, 0x63, 0x9f, 0xfa, 0xdf, 0x51, 0xe2, 0x2f, 0x42, 0x33, 0x2f, 0xf4, 0x45, 0xe8,
-	0x8f, 0x34, 0xc8, 0xee, 0xae, 0x6d, 0x20, 0x1b, 0x4a, 0xc1, 0x05, 0x1b, 0x35, 0x7d, 0x1b, 0x93,
-	0x67, 0x5c, 0xa5, 0x8d, 0x82, 0x4f, 0x1c, 0xea, 0x40, 0xfb, 0x30, 0xb5, 0xe7, 0x19, 0x26, 0x33,
-	0x2c, 0x91, 0x92, 0x9a, 0x24, 0x28, 0xf2, 0x1f, 0x82, 0xaa, 0xf3, 0x3d, 0x89, 0x8a, 0x7d, 0xf8,
-	0xda, 0x37, 0x41, 0xb9, 0x2d, 0xbe, 0xd9, 0xbd, 0x88, 0x46, 0x06, 0x9b, 0xdd, 0xb4, 0x86, 0xd6,
-	0x7e, 0x90, 0x81, 0x82, 0x1a, 0x0a, 0x17, 0x9f, 0xae, 0xa0, 0xb1, 0x74, 0xc5, 0xea, 0x84, 0x7f,
-	0x1d, 0x31, 0x32, 0x59, 0xd1, 0x4f, 0x24, 0x2b, 0x26, 0xfd, 0x8f, 0x8a, 0xe7, 0xa4, 0x2a, 0xfe,
-	0x47, 0x83, 0xe9, 0xe8, 0x9f, 0x59, 0xfc, 0x0c, 0x25, 0x2a, 0x3e, 0xd1, 0x00, 0xfc, 0xa6, 0x5f,
-	0x78, 0x9a, 0xa2, 0x1d, 0x4f, 0x53, 0xdc, 0x99, 0xb0, 0x57, 0x47, 0x24, 0x29, 0xbe, 0x95, 0xf3,
-	0x9b, 0x24, 0x52, 0x14, 0x1f, 0x69, 0x70, 0x89, 0xc4, 0xc2, 0x7e, 0xd5, 0xb2, 0x73, 0xcb, 0x22,
-	0x5c, 0x55, 0xd5, 0x48, 0xfc, 0x6f, 0x15, 0x4e, 0xa8, 0x45, 0xb7, 0x61, 0x7a, 0xa0, 0x82, 0x3f,
-	0x11, 0x02, 0x64, 0xe2, 0x17, 0x31, 0xb6, 0x23, 0x34, 0x1c, 0xe3, 0x7c, 0x4e, 0x9a, 0x25, 0x7b,
-	0x2e, 0x69, 0x96, 0xe8, 0xc1, 0x64, 0xee, 0xc4, 0x83, 0xc9, 0x5b, 0x30, 0xdd, 0x71, 0xec, 0xbe,
-	0x9f, 0x33, 0x11, 0x7f, 0x69, 0xa1, 0xae, 0xeb, 0x6c, 0x44, 0xca, 0x71, 0x8c, 0x0b, 0x79, 0x00,
-	0xcc, 0x0e, 0x64, 0x0a, 0x13, 0x26, 0xaa, 0xfc, 0xe5, 0x2f, 0x72, 0x1f, 0x25, 0x00, 0xc7, 0x11,
-	0x45, 0xb5, 0x7f, 0xcd, 0xf8, 0x73, 0xbb, 0x95, 0xb8, 0xac, 0xab, 0x8d, 0xb8, 0xac, 0xab, 0xde,
-	0x5b, 0x44, 0xc3, 0xf3, 0x37, 0xa1, 0xe0, 0x50, 0xe2, 0xda, 0x96, 0x7a, 0xb9, 0x14, 0x38, 0x12,
-	0x2c, 0x4a, 0xb1, 0xa2, 0x46, 0xc3, 0xf8, 0xcc, 0x73, 0xc2, 0xf8, 0xcf, 0x45, 0xcc, 0x9d, 0x15,
-	0x53, 0x36, 0x98, 0x39, 0x29, 0x26, 0x17, 0x11, 0x8a, 0x3a, 0x26, 0xcb, 0x27, 0x23, 0x14, 0x75,
-	0x94, 0x15, 0x70, 0xa0, 0x36, 0x4c, 0xf3, 0x70, 0x5e, 0x44, 0x29, 0xed, 0x15, 0x36, 0x46, 0x8e,
-	0x20, 0x18, 0x94, 0xcd, 0x08, 0x0e, 0x8e, 0xa1, 0xd6, 0xbe, 0x0c, 0x61, 0xf6, 0x05, 0x2d, 0x41,
-	0x69, 0xe0, 0xd8, 0x03, 0xd2, 0x25, 0x8c, 0xaa, 0xdd, 0x5d, 0xb0, 0x6c, 0x6d, 0xfb, 0x04, 0x1c,
-	0xf2, 0x34, 0xea, 0x1f, 0x7f, 0xb6, 0xf8, 0xca, 0x27, 0x9f, 0x2d, 0xbe, 0xf2, 0xe9, 0x67, 0x8b,
-	0xaf, 0xfc, 0xfe, 0xf1, 0xa2, 0xf6, 0xf1, 0xf1, 0xa2, 0xf6, 0xc9, 0xf1, 0xa2, 0xf6, 0xe9, 0xf1,
-	0xa2, 0xf6, 0xa3, 0xe3, 0x45, 0xed, 0x3b, 0xff, 0xbe, 0xf8, 0xca, 0xd7, 0x8a, 0x7e, 0x87, 0xff,
-	0x7f, 0x00, 0x00, 0x00, 0xff, 0xff, 0x0a, 0xc2, 0x59, 0xdc, 0xc2, 0x4f, 0x00, 0x00,
+	// 4453 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xec, 0x5c, 0x4f, 0x6c, 0x1c, 0x59,
+	0x5a, 0x4f, 0xf5, 0x1f, 0xbb, 0xfb, 0x6b, 0x3b, 0xb6, 0x5f, 0x32, 0xa1, 0x63, 0x32, 0xee, 0x50,
+	0xa3, 0x19, 0x05, 0xd8, 0x6d, 0x33, 0x66, 0x96, 0xcd, 0xc2, 0xce, 0x66, 0xdc, 0x76, 0xec, 0x71,
+	0x62, 0x67, 0xcc, 0x6b, 0x3b, 0x61, 0x76, 0x56, 0x84, 0xe7, 0xea, 0xd7, 0xed, 0x9a, 0xae, 0xae,
+	0xea, 0xad, 0x7a, 0xe5, 0xc4, 0x23, 0x56, 0x20, 0x71, 0x18, 0x10, 0xa0, 0x5d, 0x89, 0x0b, 0xd2,
+	0x0a, 0xc4, 0x61, 0x25, 0xc4, 0x81, 0x0b, 0x02, 0x09, 0x81, 0x56, 0xe2, 0x84, 0x86, 0xdb, 0x1c,
+	0x90, 0x18, 0xa4, 0x95, 0xb5, 0x63, 0x24, 0x6e, 0x88, 0x45, 0x73, 0x8b, 0x90, 0x40, 0xef, 0x4f,
+	0xfd, 0xed, 0x6a, 0xc7, 0xee, 0xb6, 0xc3, 0x81, 0xdc, 0xba, 0xde, 0xf7, 0xbd, 0xdf, 0xf7, 0xfe,
+	0x7e, 0xef, 0xfb, 0xf3, 0x5e, 0xc3, 0x7a, 0xc7, 0x64, 0xfb, 0xfe, 0x5e, 0xdd, 0x70, 0x7a, 0x8b,
+	0xb6, 0xdf, 0x23, 0x7d, 0xd7, 0xf9, 0x50, 0xfc, 0x68, 0x5b, 0xce, 0x93, 0xc5, 0x7e, 0xb7, 0xb3,
+	0x48, 0xfa, 0xa6, 0x17, 0x95, 0x1c, 0xbc, 0x49, 0xac, 0xfe, 0x3e, 0x79, 0x73, 0xb1, 0x43, 0x6d,
+	0xea, 0x12, 0x46, 0x5b, 0xf5, 0xbe, 0xeb, 0x30, 0x07, 0x7d, 0x35, 0x02, 0xaa, 0x07, 0x40, 0xf5,
+	0xa0, 0x5a, 0xbd, 0xdf, 0xed, 0xd4, 0x39, 0x50, 0x54, 0x12, 0x00, 0xcd, 0x7f, 0x39, 0xd6, 0x82,
+	0x8e, 0xd3, 0x71, 0x16, 0x05, 0xde, 0x9e, 0xdf, 0x16, 0x5f, 0xe2, 0x43, 0xfc, 0x92, 0x72, 0xe6,
+	0xf5, 0xee, 0x6d, 0xaf, 0x6e, 0x3a, 0xbc, 0x59, 0x8b, 0x86, 0xe3, 0xd2, 0xc5, 0x83, 0x81, 0xb6,
+	0xcc, 0xbf, 0x15, 0xf1, 0xf4, 0x88, 0xb1, 0x6f, 0xda, 0xd4, 0x3d, 0x0c, 0xfa, 0xb2, 0xe8, 0x52,
+	0xcf, 0xf1, 0x5d, 0x83, 0x9e, 0xa9, 0x96, 0xb7, 0xd8, 0xa3, 0x8c, 0x64, 0xc9, 0x5a, 0x1c, 0x56,
+	0xcb, 0xf5, 0x6d, 0x66, 0xf6, 0x06, 0xc5, 0xfc, 0xd2, 0xf3, 0x2a, 0x78, 0xc6, 0x3e, 0xed, 0x91,
+	0x74, 0x3d, 0xfd, 0x3f, 0xa7, 0xe0, 0xf2, 0xf2, 0x9e, 0xc7, 0x5c, 0x62, 0xb0, 0x87, 0xd4, 0x65,
+	0xf4, 0x29, 0xba, 0x09, 0x05, 0x9b, 0xf4, 0x68, 0x55, 0xbb, 0xa9, 0xdd, 0x2a, 0x37, 0xa6, 0x3e,
+	0x39, 0xaa, 0x5d, 0x3a, 0x3e, 0xaa, 0x15, 0x1e, 0x90, 0x1e, 0xc5, 0x82, 0x82, 0x0c, 0x98, 0x90,
+	0xbd, 0xad, 0xe6, 0x6f, 0x6a, 0xb7, 0x2a, 0x4b, 0x77, 0xea, 0x23, 0x4e, 0x53, 0xbd, 0x29, 0x60,
+	0x1a, 0x70, 0x7c, 0x54, 0x9b, 0x90, 0xbf, 0xb1, 0x82, 0x46, 0x1f, 0x40, 0xc1, 0x33, 0xed, 0x6e,
+	0xb5, 0x20, 0x44, 0xbc, 0x3d, 0xba, 0x08, 0xd3, 0xee, 0x36, 0x4a, 0xbc, 0x07, 0xfc, 0x17, 0x16,
+	0xa0, 0xe8, 0xbb, 0x1a, 0xcc, 0x19, 0x8e, 0xcd, 0x08, 0x1f, 0xa8, 0x1d, 0xda, 0xeb, 0x5b, 0x84,
+	0xd1, 0x6a, 0x51, 0x88, 0xba, 0x37, 0xb2, 0xa8, 0x95, 0x34, 0x62, 0xe3, 0x95, 0xe3, 0xa3, 0xda,
+	0xdc, 0x40, 0x31, 0x1e, 0x94, 0x8d, 0x1e, 0x41, 0xde, 0x6f, 0xb5, 0xab, 0x13, 0xa2, 0x09, 0x5f,
+	0x1f, 0xb9, 0x09, 0xbb, 0xab, 0x6b, 0x8d, 0xc9, 0xe3, 0xa3, 0x5a, 0x7e, 0x77, 0x75, 0x0d, 0x73,
+	0x44, 0xd4, 0x85, 0x12, 0x5f, 0x65, 0x2d, 0xc2, 0x48, 0x75, 0x52, 0xa0, 0x2f, 0x8f, 0x8c, 0xbe,
+	0xa5, 0x80, 0x1a, 0x53, 0xc7, 0x47, 0xb5, 0x52, 0xf0, 0x85, 0x43, 0x01, 0xe8, 0x8f, 0x34, 0x98,
+	0xb2, 0x9d, 0x16, 0x6d, 0x52, 0x8b, 0x1a, 0xcc, 0x71, 0xab, 0xa5, 0x9b, 0xf9, 0x5b, 0x95, 0xa5,
+	0xf7, 0x47, 0x96, 0x98, 0x5c, 0x9b, 0xf5, 0x07, 0x31, 0xec, 0xbb, 0x36, 0x73, 0x0f, 0x1b, 0x57,
+	0xd5, 0xfa, 0x9c, 0x8a, 0x93, 0x70, 0xa2, 0x11, 0x68, 0x17, 0x2a, 0xcc, 0xb1, 0xf8, 0xba, 0x37,
+	0x1d, 0xdb, 0xab, 0x96, 0x45, 0x9b, 0x16, 0xea, 0x72, 0xcb, 0x70, 0xc9, 0x75, 0xbe, 0xe7, 0xeb,
+	0x07, 0x6f, 0xd6, 0x77, 0x42, 0xb6, 0xc6, 0x15, 0x05, 0x5c, 0x89, 0xca, 0x3c, 0x1c, 0xc7, 0x41,
+	0x14, 0x66, 0x3c, 0x6a, 0xf8, 0xae, 0xc9, 0x0e, 0xf9, 0x14, 0xd3, 0xa7, 0xac, 0x0a, 0x62, 0x80,
+	0xdf, 0xc8, 0x82, 0xde, 0x76, 0x5a, 0xcd, 0x24, 0x77, 0xe3, 0xca, 0xf1, 0x51, 0x6d, 0x26, 0x55,
+	0x88, 0xd3, 0x98, 0xc8, 0x86, 0x59, 0xb3, 0x47, 0x3a, 0x74, 0xdb, 0xb7, 0xac, 0x26, 0x35, 0x5c,
+	0xca, 0xbc, 0x6a, 0x45, 0x74, 0xe1, 0x56, 0x96, 0x9c, 0x4d, 0xc7, 0x20, 0xd6, 0x7b, 0x7b, 0x1f,
+	0x52, 0x83, 0x61, 0xda, 0xa6, 0x2e, 0xb5, 0x0d, 0xda, 0xa8, 0xaa, 0xce, 0xcc, 0x6e, 0xa4, 0x90,
+	0xf0, 0x00, 0x36, 0x5a, 0x87, 0xb9, 0xbe, 0x6b, 0x3a, 0xa2, 0x09, 0x16, 0xf1, 0x3c, 0xbe, 0xf1,
+	0xab, 0x53, 0x42, 0x19, 0x5c, 0x57, 0x30, 0x73, 0xdb, 0x69, 0x06, 0x3c, 0x58, 0x07, 0xdd, 0x82,
+	0x52, 0x50, 0x58, 0x9d, 0xbe, 0xa9, 0xdd, 0x2a, 0xca, 0x65, 0x13, 0xd4, 0xc5, 0x21, 0x15, 0xad,
+	0x41, 0x89, 0xb4, 0xdb, 0xa6, 0xcd, 0x39, 0x2f, 0x8b, 0x21, 0xbc, 0x91, 0xd5, 0xb5, 0x65, 0xc5,
+	0x23, 0x71, 0x82, 0x2f, 0x1c, 0xd6, 0x45, 0xf7, 0x00, 0x79, 0xd4, 0x3d, 0x30, 0x0d, 0xba, 0x6c,
+	0x18, 0x8e, 0x6f, 0x33, 0xd1, 0xf6, 0x19, 0xd1, 0xf6, 0x79, 0xd5, 0x76, 0xd4, 0x1c, 0xe0, 0xc0,
+	0x19, 0xb5, 0xd0, 0x5d, 0x98, 0x3c, 0x70, 0x2c, 0xbf, 0x47, 0xbd, 0xea, 0xac, 0x18, 0xed, 0xf9,
+	0xac, 0x26, 0x3d, 0x14, 0x2c, 0x8d, 0x19, 0x05, 0x3e, 0x29, 0xbf, 0x3d, 0x1c, 0xd4, 0x45, 0x26,
+	0x4c, 0x58, 0x66, 0xcf, 0x64, 0x5e, 0x75, 0x4e, 0x74, 0xec, 0xee, 0xc8, 0x5b, 0x41, 0x6e, 0x81,
+	0x4d, 0x01, 0x26, 0x35, 0xa6, 0xfc, 0x8d, 0x95, 0x00, 0x64, 0x40, 0xd1, 0x33, 0x88, 0x45, 0xab,
+	0x48, 0x48, 0xfa, 0xc6, 0xe8, 0x2a, 0x93, 0xa3, 0x34, 0xa6, 0x55, 0x9f, 0x8a, 0xe2, 0x13, 0x4b,
+	0xec, 0xf9, 0x3b, 0x30, 0x37, 0xb0, 0x09, 0xd1, 0x2c, 0xe4, 0xbb, 0xf4, 0x50, 0x9e, 0x18, 0x98,
+	0xff, 0x44, 0x57, 0xa1, 0x78, 0x40, 0x2c, 0x9f, 0x56, 0x73, 0xa2, 0x4c, 0x7e, 0xfc, 0x72, 0xee,
+	0xb6, 0xa6, 0x3f, 0x82, 0xe9, 0x65, 0x9f, 0xed, 0x3b, 0xae, 0xf9, 0x91, 0xd8, 0x47, 0x68, 0x0d,
+	0x8a, 0xcc, 0xe9, 0x52, 0x5b, 0x54, 0xaf, 0x2c, 0xbd, 0x9e, 0x35, 0xcc, 0x72, 0x6d, 0xde, 0xa7,
+	0x87, 0x81, 0xdc, 0x46, 0x99, 0xb7, 0x6c, 0x87, 0xd7, 0xc3, 0xb2, 0xba, 0xfe, 0x4d, 0x98, 0x68,
+	0xf8, 0xed, 0x36, 0x75, 0x4f, 0x71, 0x82, 0xd5, 0xa1, 0xc0, 0x0e, 0xfb, 0xaa, 0x75, 0xe1, 0xd2,
+	0x28, 0xec, 0x1c, 0xf6, 0xe9, 0xb3, 0xa3, 0x1a, 0x48, 0x1c, 0xfe, 0x85, 0x05, 0x9f, 0xfe, 0x85,
+	0x06, 0x57, 0x64, 0xa1, 0x5a, 0x3d, 0x2b, 0x8e, 0xdd, 0x36, 0x3b, 0x88, 0x42, 0xd1, 0xa5, 0x2d,
+	0xd3, 0x53, 0x6d, 0x5f, 0x1d, 0x79, 0xc8, 0x31, 0x47, 0x91, 0xa0, 0xb2, 0x6b, 0xa2, 0x00, 0x4b,
+	0x74, 0xe4, 0x43, 0xf9, 0x43, 0xca, 0x3c, 0xe6, 0x52, 0xd2, 0x13, 0x6d, 0xae, 0x2c, 0xbd, 0x3b,
+	0xb2, 0xa8, 0x7b, 0x94, 0x35, 0x05, 0x92, 0x12, 0x37, 0x7d, 0x7c, 0x54, 0x2b, 0x87, 0x85, 0x38,
+	0x92, 0xa4, 0xff, 0x7b, 0x0e, 0xca, 0xe1, 0xe1, 0x85, 0x5e, 0x83, 0xa2, 0xd0, 0x15, 0x6a, 0x58,
+	0xc3, 0xe5, 0x21, 0x54, 0x0a, 0x96, 0x34, 0xf4, 0x3a, 0x4c, 0x1a, 0x4e, 0xaf, 0x47, 0xec, 0x56,
+	0x35, 0x77, 0x33, 0x7f, 0xab, 0xdc, 0xa8, 0xf0, 0x5d, 0xb1, 0x22, 0x8b, 0x70, 0x40, 0x43, 0x37,
+	0xa0, 0x40, 0xdc, 0x8e, 0x57, 0xcd, 0x0b, 0x1e, 0x71, 0x3a, 0x2f, 0xbb, 0x1d, 0x0f, 0x8b, 0x52,
+	0xf4, 0x35, 0xc8, 0x53, 0xfb, 0xa0, 0x5a, 0x18, 0xbe, 0xed, 0xee, 0xda, 0x07, 0x0f, 0x89, 0xdb,
+	0xa8, 0xa8, 0x36, 0xe4, 0xef, 0xda, 0x07, 0x98, 0xd7, 0x41, 0xef, 0xc3, 0x94, 0xdc, 0x79, 0x5b,
+	0x7c, 0x23, 0x7b, 0xd5, 0xa2, 0xc0, 0xa8, 0x0d, 0xdf, 0xba, 0x82, 0x2f, 0x3a, 0x45, 0x62, 0x85,
+	0x1e, 0x4e, 0x40, 0xa1, 0xf7, 0xa1, 0x1c, 0x58, 0x79, 0x9e, 0x3a, 0xa7, 0x33, 0x15, 0x30, 0x56,
+	0x4c, 0x98, 0x7e, 0xdb, 0x37, 0x5d, 0xda, 0xa3, 0x36, 0xf3, 0x1a, 0x73, 0x4a, 0x40, 0x39, 0xa0,
+	0x7a, 0x38, 0x42, 0xd3, 0xff, 0x2b, 0x07, 0x83, 0x56, 0x42, 0x52, 0xa0, 0x76, 0x9e, 0x02, 0xd1,
+	0x1e, 0xcc, 0x84, 0x7a, 0x7f, 0xdb, 0xb1, 0x4c, 0xe3, 0x50, 0x6d, 0x85, 0xdb, 0xaa, 0xda, 0xcc,
+	0x46, 0x92, 0xfc, 0xec, 0xa8, 0xf6, 0xea, 0xa0, 0x8d, 0x5c, 0x8f, 0x18, 0x70, 0x1a, 0x90, 0xcb,
+	0x48, 0x1f, 0x8f, 0xd2, 0x5c, 0x7c, 0x6d, 0xc8, 0x0e, 0x1f, 0xe1, 0x6c, 0x1c, 0x7d, 0xa5, 0xe8,
+	0x7f, 0x91, 0x83, 0xc2, 0xdd, 0x56, 0x87, 0x72, 0x6d, 0xd1, 0x76, 0x9d, 0x5e, 0x5a, 0x5b, 0xac,
+	0xb9, 0x4e, 0x0f, 0x0b, 0x0a, 0x9a, 0x87, 0x1c, 0x73, 0xd4, 0x00, 0x81, 0xa2, 0xe7, 0x76, 0x1c,
+	0x9c, 0x63, 0x0e, 0xfa, 0x08, 0xc0, 0x70, 0xec, 0x96, 0x29, 0x4d, 0x8b, 0xfc, 0x98, 0x16, 0xe4,
+	0x9a, 0xe3, 0x3e, 0x21, 0x6e, 0x6b, 0x25, 0x44, 0x6c, 0x5c, 0x3e, 0x3e, 0xaa, 0x41, 0xf4, 0x8d,
+	0x63, 0xd2, 0x50, 0x27, 0x3c, 0x5b, 0xa4, 0x91, 0xbc, 0x32, 0xb2, 0x5c, 0x3e, 0x10, 0xc3, 0x4f,
+	0x16, 0xfd, 0x0f, 0x35, 0x80, 0x88, 0x05, 0xbd, 0x0d, 0x33, 0x7b, 0x42, 0x19, 0x6e, 0x91, 0xa7,
+	0x9b, 0xd4, 0xee, 0xb0, 0x7d, 0x31, 0x78, 0x05, 0x39, 0x69, 0x8d, 0x24, 0x09, 0xa7, 0x79, 0xd1,
+	0x3b, 0x30, 0x2b, 0x8b, 0x76, 0x3d, 0xa2, 0x30, 0xc5, 0xe0, 0x4e, 0x37, 0xae, 0x72, 0x13, 0xa5,
+	0x91, 0xa2, 0xe1, 0x01, 0x6e, 0xfd, 0x2d, 0x98, 0x1b, 0x18, 0x29, 0x54, 0x83, 0x62, 0x97, 0x1e,
+	0x6e, 0xf0, 0x73, 0x84, 0x2b, 0x15, 0xa1, 0x45, 0xef, 0xf3, 0x02, 0x2c, 0xcb, 0xf5, 0xff, 0xd6,
+	0xa0, 0xb4, 0xe6, 0xdb, 0x86, 0x38, 0x75, 0x9e, 0x7f, 0x46, 0x04, 0x3a, 0x2a, 0x97, 0xa9, 0xa3,
+	0x7c, 0x98, 0xe8, 0x3e, 0x09, 0x75, 0x58, 0x65, 0x69, 0x6b, 0xf4, 0x39, 0x57, 0x4d, 0xaa, 0xdf,
+	0x17, 0x78, 0xd2, 0xac, 0xbd, 0xac, 0x1a, 0x34, 0x71, 0xff, 0x91, 0x10, 0xaa, 0x84, 0xcd, 0x7f,
+	0x0d, 0x2a, 0x31, 0xb6, 0x33, 0x1d, 0xbc, 0x7f, 0xa9, 0xc1, 0xcc, 0xba, 0x74, 0xff, 0x1c, 0x57,
+	0x3a, 0x5b, 0xe8, 0x3a, 0xe4, 0xdd, 0xbe, 0x2f, 0xea, 0xe7, 0xa5, 0xdf, 0x80, 0xb7, 0x77, 0x31,
+	0x2f, 0x43, 0xbf, 0x06, 0xa5, 0x96, 0x2f, 0x4d, 0x5d, 0x75, 0xe4, 0xd4, 0x63, 0xfb, 0x2b, 0x74,
+	0x32, 0xa3, 0x9e, 0x71, 0x27, 0x80, 0xef, 0xb8, 0x55, 0x55, 0x4b, 0x5a, 0x69, 0xc1, 0x17, 0x0e,
+	0xd1, 0xf8, 0x19, 0xd1, 0xf3, 0x3a, 0x4d, 0xf3, 0x23, 0xe9, 0x3f, 0x16, 0xe5, 0x19, 0xb1, 0x25,
+	0x8b, 0x70, 0x40, 0xd3, 0xbf, 0x9b, 0x83, 0x6b, 0xeb, 0x94, 0xad, 0x12, 0xda, 0x73, 0xec, 0x55,
+	0xda, 0xb7, 0x9c, 0x43, 0xae, 0xda, 0x30, 0xfd, 0x36, 0x7a, 0x07, 0xc0, 0xf4, 0xf6, 0x9a, 0x07,
+	0x06, 0x3f, 0xa2, 0xd5, 0x14, 0xde, 0x54, 0x23, 0x06, 0x1b, 0xcd, 0x86, 0xa2, 0x3c, 0x4b, 0x7c,
+	0xe1, 0x58, 0x9d, 0xe8, 0x30, 0xcb, 0x9d, 0x70, 0x98, 0x35, 0x01, 0xfa, 0x91, 0x82, 0xcc, 0x0b,
+	0xce, 0x5f, 0x0c, 0xc4, 0x9c, 0x45, 0x37, 0xc6, 0x60, 0xc6, 0x51, 0x59, 0x7f, 0x97, 0x87, 0xf9,
+	0x75, 0xca, 0xc2, 0xb3, 0x5a, 0xd9, 0x22, 0xcd, 0x3e, 0x35, 0xf8, 0xa8, 0x7c, 0xac, 0xc1, 0x84,
+	0x45, 0xf6, 0xa8, 0xe5, 0x89, 0x2d, 0x50, 0x59, 0x7a, 0x3c, 0xf2, 0x9a, 0x1c, 0x2e, 0xa5, 0xbe,
+	0x29, 0x24, 0xa4, 0x56, 0xa9, 0x2c, 0xc4, 0x4a, 0x3c, 0xfa, 0x0a, 0x54, 0x0c, 0xcb, 0xf7, 0x18,
+	0x75, 0xb7, 0x1d, 0x57, 0x6e, 0xee, 0x62, 0xe4, 0x50, 0xad, 0x44, 0x24, 0x1c, 0xe7, 0x43, 0x4b,
+	0x00, 0x86, 0x65, 0x52, 0x9b, 0x89, 0x5a, 0x72, 0x6d, 0xa0, 0x60, 0xbc, 0x57, 0x42, 0x0a, 0x8e,
+	0x71, 0x71, 0x51, 0x3d, 0xc7, 0x36, 0x99, 0x23, 0x45, 0x15, 0x92, 0xa2, 0xb6, 0x22, 0x12, 0x8e,
+	0xf3, 0x89, 0x6a, 0x94, 0xb9, 0xa6, 0xe1, 0x89, 0x6a, 0xc5, 0x54, 0xb5, 0x88, 0x84, 0xe3, 0x7c,
+	0x7c, 0xfb, 0xc5, 0xfa, 0x7f, 0xa6, 0xed, 0xf7, 0xf7, 0x25, 0x58, 0x48, 0x0c, 0x2b, 0x23, 0x8c,
+	0xb6, 0x7d, 0xab, 0x49, 0x59, 0x30, 0x81, 0x5f, 0x81, 0x8a, 0x72, 0x44, 0x1e, 0x44, 0xaa, 0x29,
+	0x6c, 0x54, 0x33, 0x22, 0xe1, 0x38, 0x1f, 0xfa, 0xfd, 0x68, 0xde, 0x73, 0x62, 0xde, 0x8d, 0xf3,
+	0x99, 0xf7, 0x81, 0x06, 0x9e, 0x6a, 0xee, 0x17, 0xa1, 0x6c, 0x13, 0xe6, 0x89, 0x8d, 0xa4, 0xf6,
+	0x4c, 0x68, 0x8b, 0x3c, 0x08, 0x08, 0x38, 0xe2, 0x41, 0xdb, 0x70, 0x55, 0x0d, 0xf1, 0xdd, 0xa7,
+	0x7d, 0xc7, 0x65, 0xd4, 0x95, 0x75, 0x0b, 0xa2, 0xee, 0x0d, 0x55, 0xf7, 0xea, 0x56, 0x06, 0x0f,
+	0xce, 0xac, 0x89, 0xb6, 0xe0, 0x8a, 0x21, 0x6c, 0x5b, 0x4c, 0x2d, 0x87, 0xb4, 0x02, 0xc0, 0xa2,
+	0x00, 0xfc, 0x69, 0x05, 0x78, 0x65, 0x65, 0x90, 0x05, 0x67, 0xd5, 0x4b, 0xaf, 0xe6, 0x89, 0x91,
+	0x56, 0xf3, 0xe4, 0x28, 0xab, 0xb9, 0x34, 0xda, 0x6a, 0x2e, 0x9f, 0x6e, 0x35, 0xf3, 0x91, 0xe7,
+	0xeb, 0x88, 0xba, 0xdc, 0x21, 0x93, 0x2e, 0x96, 0x58, 0x78, 0x90, 0x1c, 0xf9, 0x66, 0x06, 0x0f,
+	0xce, 0xac, 0x89, 0xf6, 0x60, 0x5e, 0x96, 0xdf, 0xb5, 0x0d, 0xf7, 0xb0, 0xcf, 0xd5, 0x7d, 0x0c,
+	0xb7, 0x22, 0x70, 0x75, 0x85, 0x3b, 0xdf, 0x1c, 0xca, 0x89, 0x4f, 0x40, 0x41, 0xbf, 0x02, 0xd3,
+	0x72, 0x96, 0xb6, 0x48, 0x3f, 0x16, 0x9b, 0x78, 0x45, 0xc1, 0x4e, 0xaf, 0xc4, 0x89, 0x38, 0xc9,
+	0x8b, 0x96, 0x61, 0xa6, 0x7f, 0x60, 0xf0, 0x9f, 0x1b, 0xed, 0x07, 0x94, 0xb6, 0x68, 0x4b, 0x84,
+	0x26, 0xca, 0x8d, 0x9f, 0x0a, 0x0c, 0xdf, 0xed, 0x24, 0x19, 0xa7, 0xf9, 0xd1, 0x6d, 0x98, 0xf2,
+	0x18, 0x71, 0x99, 0x72, 0x6a, 0x44, 0xc0, 0xa2, 0x1c, 0x79, 0x10, 0xcd, 0x18, 0x0d, 0x27, 0x38,
+	0xc7, 0xd1, 0x1e, 0xcf, 0xe4, 0x61, 0x28, 0xbc, 0xc2, 0x94, 0xda, 0xff, 0x9d, 0xb4, 0xda, 0xff,
+	0x60, 0x9c, 0xed, 0x9f, 0x21, 0xe1, 0x54, 0xdb, 0xfe, 0x1e, 0x20, 0x57, 0xf9, 0xb0, 0xd2, 0x8d,
+	0x89, 0x69, 0xfe, 0x30, 0xf4, 0x82, 0x07, 0x38, 0x70, 0x46, 0x2d, 0xd4, 0x84, 0x57, 0x3c, 0x6a,
+	0x33, 0xd3, 0xa6, 0x56, 0x12, 0x4e, 0x1e, 0x09, 0xaf, 0x2a, 0xb8, 0x57, 0x9a, 0x59, 0x4c, 0x38,
+	0xbb, 0xee, 0x38, 0x83, 0xff, 0xa3, 0xb2, 0x38, 0x77, 0xe5, 0xd0, 0x9c, 0x9b, 0xda, 0xfe, 0x38,
+	0xad, 0xb6, 0x1f, 0x8f, 0x3f, 0x6f, 0xa3, 0xa9, 0xec, 0x25, 0x00, 0x31, 0x0b, 0x71, 0x9d, 0x1d,
+	0x6a, 0x2a, 0x1c, 0x52, 0x70, 0x8c, 0x8b, 0xef, 0xc2, 0x60, 0x9c, 0xe3, 0xea, 0x3a, 0xdc, 0x85,
+	0xcd, 0x38, 0x11, 0x27, 0x79, 0x87, 0xaa, 0xfc, 0xe2, 0xc8, 0x2a, 0xff, 0x1e, 0x20, 0xd3, 0x36,
+	0x59, 0x38, 0xe5, 0x12, 0x6f, 0x22, 0x19, 0xf9, 0xdb, 0x18, 0xe0, 0xc0, 0x19, 0xb5, 0x86, 0x2c,
+	0xe5, 0xc9, 0xf3, 0x5d, 0xca, 0xa5, 0xd1, 0x97, 0x32, 0x7a, 0x0c, 0xd7, 0x85, 0x28, 0x35, 0x3e,
+	0x49, 0x60, 0xa9, 0xfc, 0x7f, 0x46, 0x01, 0x5f, 0xc7, 0xc3, 0x18, 0xf1, 0x70, 0x0c, 0x3e, 0x3f,
+	0x86, 0x4b, 0x5b, 0x5c, 0x38, 0xb1, 0x86, 0x1f, 0x0c, 0x2b, 0x19, 0x3c, 0x38, 0xb3, 0x26, 0x5f,
+	0x62, 0x8c, 0x2f, 0x43, 0xb2, 0x67, 0xd1, 0x96, 0x38, 0x08, 0x4a, 0xd1, 0x12, 0xdb, 0xd9, 0x6c,
+	0x2a, 0x0a, 0x8e, 0x71, 0x65, 0xe9, 0xea, 0xa9, 0x33, 0xea, 0xea, 0x75, 0x91, 0xe6, 0x69, 0x27,
+	0x8e, 0x04, 0xa5, 0xf0, 0xc3, 0x58, 0xf6, 0x4a, 0x9a, 0x01, 0x0f, 0xd6, 0x11, 0x47, 0xa5, 0xe1,
+	0x9a, 0x7d, 0xe6, 0x25, 0xb1, 0x2e, 0xa7, 0x8e, 0xca, 0x0c, 0x1e, 0x9c, 0x59, 0x93, 0x1b, 0x29,
+	0xfb, 0x94, 0x58, 0x6c, 0x3f, 0x09, 0x38, 0x93, 0x34, 0x52, 0xde, 0x1d, 0x64, 0xc1, 0x59, 0xf5,
+	0xc6, 0x51, 0x6f, 0x7f, 0x90, 0x83, 0x2b, 0xeb, 0x54, 0xa5, 0x58, 0xb6, 0x9d, 0x56, 0xa0, 0xd7,
+	0xfe, 0x9f, 0x7a, 0x59, 0x7f, 0xa2, 0x01, 0xbc, 0xbb, 0xb3, 0xb3, 0xad, 0x5c, 0xe4, 0x16, 0x14,
+	0x88, 0xaf, 0x22, 0x1c, 0x95, 0xa5, 0xb5, 0xd1, 0x33, 0x59, 0xf1, 0xa0, 0xb7, 0x0a, 0x27, 0xf8,
+	0x6c, 0x1f, 0x0b, 0x74, 0xf4, 0xb3, 0x30, 0xa9, 0xce, 0x06, 0x31, 0x56, 0xa5, 0x28, 0xa3, 0xa0,
+	0xce, 0x0f, 0x1c, 0xd0, 0xf5, 0x9f, 0xe4, 0xe0, 0xda, 0x86, 0xcd, 0xa8, 0xdb, 0x64, 0xb4, 0x9f,
+	0x08, 0x4a, 0xa3, 0xdf, 0x88, 0xe5, 0xfa, 0x64, 0x7b, 0x7f, 0xe1, 0x74, 0x3e, 0xbb, 0xcc, 0x17,
+	0x6d, 0x51, 0x46, 0xa2, 0x5d, 0x19, 0x95, 0xc5, 0x12, 0x7c, 0x3e, 0x14, 0xbc, 0x3e, 0x35, 0x54,
+	0x44, 0xa0, 0x39, 0xf2, 0x68, 0x64, 0x77, 0x80, 0xaf, 0xbc, 0x28, 0x16, 0x23, 0xd6, 0xa1, 0x10,
+	0x87, 0xbe, 0x03, 0x13, 0x1e, 0x23, 0xcc, 0x0f, 0x22, 0x6c, 0xbb, 0xe7, 0x2d, 0x58, 0x80, 0x47,
+	0x07, 0xa4, 0xfc, 0xc6, 0x4a, 0xa8, 0xfe, 0x13, 0x0d, 0xe6, 0xb3, 0x2b, 0x6e, 0x9a, 0x1e, 0x43,
+	0xdf, 0x1a, 0x18, 0xf6, 0x53, 0x86, 0x4a, 0x78, 0x6d, 0x31, 0xe8, 0xb3, 0x4a, 0x70, 0x29, 0x28,
+	0x89, 0x0d, 0x39, 0x83, 0xa2, 0xc9, 0x68, 0x2f, 0xb0, 0x12, 0xde, 0x3b, 0xe7, 0xae, 0xc7, 0x76,
+	0x25, 0x97, 0x82, 0xa5, 0x30, 0xfd, 0xe3, 0xdc, 0xb0, 0x2e, 0xf3, 0x69, 0x41, 0xdd, 0x64, 0xe2,
+	0xe3, 0xde, 0x78, 0x89, 0x8f, 0x86, 0x1f, 0x6b, 0xcf, 0x60, 0xfa, 0xe3, 0x37, 0x07, 0xd3, 0x1f,
+	0xef, 0x8d, 0x9f, 0xfe, 0x48, 0x8d, 0xc2, 0xd0, 0x2c, 0xc8, 0x8f, 0x72, 0x70, 0xe3, 0xa4, 0x55,
+	0x83, 0x3a, 0xe1, 0xe2, 0xd4, 0xc6, 0xbd, 0x0e, 0x71, 0xe2, 0x32, 0x44, 0x4b, 0x50, 0xec, 0xef,
+	0x13, 0x2f, 0x50, 0xa7, 0xc1, 0xa9, 0x53, 0xdc, 0xe6, 0x85, 0xcf, 0x8e, 0x6a, 0x15, 0xa9, 0x86,
+	0xc5, 0x27, 0x96, 0xac, 0x5c, 0xb1, 0xf4, 0xa8, 0xe7, 0x45, 0x86, 0x5d, 0xa8, 0x58, 0xb6, 0x64,
+	0x31, 0x0e, 0xe8, 0x88, 0xc1, 0x84, 0x74, 0x96, 0x54, 0x38, 0x79, 0x73, 0xe4, 0x7e, 0x64, 0xa4,
+	0xca, 0xa2, 0x4e, 0x29, 0xbf, 0x5b, 0xc9, 0xd2, 0xff, 0xea, 0x32, 0x5c, 0xcb, 0x9e, 0x13, 0xde,
+	0xf6, 0x03, 0xea, 0x7a, 0xa6, 0x63, 0xab, 0xd3, 0x27, 0x4a, 0xb3, 0xca, 0x62, 0x1c, 0xd0, 0xd1,
+	0x2d, 0x28, 0xb9, 0xb4, 0x6f, 0x99, 0x06, 0xf1, 0x94, 0xd3, 0x21, 0xa2, 0x8f, 0x58, 0x95, 0xe1,
+	0x90, 0x3a, 0xe4, 0xea, 0x47, 0xfe, 0xff, 0xf0, 0xea, 0xc7, 0x9f, 0x6b, 0xdc, 0x9e, 0x93, 0x11,
+	0x87, 0x81, 0x0a, 0x6a, 0x2e, 0xce, 0xb3, 0x65, 0xaf, 0x4a, 0xbb, 0x70, 0x88, 0x40, 0x3c, 0xbc,
+	0x2d, 0xe8, 0x07, 0x1a, 0x54, 0x7b, 0x29, 0x83, 0xf1, 0x02, 0x6f, 0xcf, 0xdc, 0x38, 0x3e, 0xaa,
+	0x55, 0xb7, 0x86, 0xc8, 0xc3, 0x43, 0x5b, 0x82, 0x7e, 0x0b, 0x2a, 0x7d, 0xbe, 0x2e, 0x3c, 0x46,
+	0x6d, 0x83, 0xaa, 0x5c, 0xdd, 0xe8, 0xab, 0x79, 0x3b, 0xc2, 0x6a, 0x32, 0x97, 0x30, 0xda, 0x39,
+	0x6c, 0xcc, 0x70, 0xd7, 0x2e, 0x46, 0xc0, 0x71, 0x89, 0x89, 0x3b, 0x37, 0x5b, 0x17, 0x7d, 0xe7,
+	0xe6, 0xfb, 0xd9, 0x77, 0x6e, 0xc8, 0x39, 0x6b, 0xc8, 0x97, 0x77, 0x6f, 0x5e, 0xde, 0xbd, 0x79,
+	0x51, 0x77, 0x6f, 0x6e, 0x41, 0xc9, 0xa3, 0x8c, 0x99, 0x76, 0xc7, 0xab, 0xce, 0xca, 0x04, 0x1d,
+	0x97, 0xda, 0x54, 0x65, 0x38, 0xa4, 0xa2, 0x9f, 0x87, 0xb2, 0x08, 0xb1, 0x2d, 0xbb, 0x1d, 0xaf,
+	0x3a, 0x27, 0x32, 0x75, 0xe2, 0x24, 0x6f, 0x06, 0x85, 0x38, 0xa2, 0xa3, 0xb7, 0x60, 0x4a, 0xa6,
+	0x12, 0xe5, 0x11, 0x24, 0xee, 0xc9, 0x94, 0x1b, 0xb3, 0x7c, 0x05, 0x37, 0x62, 0xe5, 0x38, 0xc1,
+	0xc5, 0x5d, 0x57, 0x1a, 0xc6, 0x21, 0xab, 0x57, 0x92, 0xae, 0x6b, 0x14, 0xa1, 0xc4, 0x31, 0x2e,
+	0xf4, 0x2a, 0xe4, 0x99, 0xe5, 0x55, 0xaf, 0x0a, 0xe6, 0xd0, 0xc5, 0xd8, 0xd9, 0x6c, 0x62, 0x5e,
+	0x3e, 0xfe, 0x25, 0x9a, 0xff, 0xd1, 0x60, 0x26, 0x75, 0x8f, 0x83, 0xcb, 0xf4, 0x5d, 0x4b, 0x9d,
+	0x94, 0xa1, 0xcc, 0x5d, 0xbc, 0x89, 0x79, 0x39, 0x7a, 0xac, 0xfc, 0x98, 0xdc, 0x98, 0xfa, 0xe8,
+	0xc1, 0xf2, 0x4e, 0x93, 0x3b, 0x2e, 0x03, 0x2e, 0xcc, 0xed, 0xd4, 0xe8, 0xe6, 0x93, 0x71, 0xd1,
+	0x93, 0x47, 0x38, 0x16, 0x1c, 0x28, 0x9c, 0x26, 0x38, 0xa0, 0x7f, 0xa6, 0x41, 0xf9, 0x3e, 0x69,
+	0x77, 0x49, 0xd3, 0xb4, 0xbb, 0xe8, 0x75, 0x98, 0xdc, 0x73, 0x9d, 0x2e, 0x75, 0x3d, 0x95, 0xfd,
+	0x15, 0x29, 0xc5, 0x86, 0x2c, 0xc2, 0x01, 0x8d, 0xfb, 0xa3, 0xcc, 0xe9, 0x9b, 0x46, 0xda, 0x1f,
+	0xdd, 0xe1, 0x85, 0x58, 0xd2, 0xd0, 0x23, 0x39, 0x77, 0xf9, 0x31, 0x6f, 0x62, 0xee, 0x6c, 0x36,
+	0x65, 0x46, 0x35, 0x98, 0x75, 0xf4, 0x46, 0xc2, 0xbe, 0x2a, 0x0f, 0xb5, 0x88, 0xbe, 0x9f, 0x83,
+	0x8a, 0xec, 0x9a, 0xf4, 0x40, 0xcf, 0xb3, 0x73, 0x77, 0x44, 0xf0, 0xdc, 0xf3, 0x7b, 0xd4, 0x5d,
+	0x77, 0x1d, 0xbf, 0xaf, 0x66, 0x29, 0x1e, 0x0c, 0x89, 0x88, 0x61, 0x00, 0x3d, 0x2a, 0x0a, 0x46,
+	0xa7, 0x70, 0x81, 0xa3, 0x53, 0x3c, 0x71, 0x74, 0xfe, 0x5a, 0x83, 0xf2, 0xa6, 0xd9, 0xa6, 0xc6,
+	0xa1, 0x61, 0x51, 0xf4, 0x2d, 0xa8, 0xb6, 0xa8, 0x45, 0x19, 0x5d, 0x77, 0x89, 0x41, 0xb7, 0xa9,
+	0x6b, 0x0a, 0x55, 0xef, 0xd8, 0x2d, 0x69, 0x8d, 0x17, 0xc3, 0x88, 0x45, 0x75, 0x75, 0x08, 0x1f,
+	0x1e, 0x8a, 0x80, 0x36, 0x60, 0xaa, 0x45, 0x3d, 0xd3, 0xa5, 0xad, 0xed, 0x98, 0xdd, 0xfd, 0x7a,
+	0xb0, 0xa4, 0x57, 0x63, 0xb4, 0x67, 0x47, 0xb5, 0xe9, 0x6d, 0xb3, 0x4f, 0x2d, 0xd3, 0xa6, 0xd2,
+	0x00, 0x4f, 0x54, 0xd5, 0x8b, 0x90, 0xdf, 0x74, 0x3a, 0xfa, 0xef, 0xe6, 0x21, 0x3c, 0xc3, 0xd1,
+	0xef, 0x69, 0x50, 0x21, 0xb6, 0xed, 0x30, 0x75, 0x38, 0xca, 0xf0, 0x3d, 0x1e, 0xdb, 0x54, 0xa8,
+	0x2f, 0x47, 0xa0, 0xf2, 0xa4, 0x0e, 0x0f, 0xd4, 0x18, 0x05, 0xc7, 0x65, 0x23, 0x3f, 0x15, 0x8c,
+	0xde, 0x1a, 0xbf, 0x15, 0xa7, 0x08, 0x3d, 0xcf, 0x7f, 0x03, 0x66, 0xd3, 0x8d, 0x3d, 0x8b, 0x22,
+	0x1c, 0x27, 0xec, 0xf5, 0x67, 0x1a, 0x94, 0x02, 0x65, 0x86, 0x56, 0xa0, 0xe0, 0x7b, 0xd4, 0x3d,
+	0xdb, 0x1d, 0x44, 0xa1, 0x01, 0x77, 0x3d, 0xea, 0x62, 0x51, 0x19, 0xbd, 0x07, 0xa5, 0x3e, 0xf1,
+	0xbc, 0x27, 0x8e, 0xdb, 0x52, 0x6a, 0xf6, 0x94, 0x40, 0xf2, 0x6c, 0x56, 0x55, 0x71, 0x08, 0xa2,
+	0xff, 0x70, 0x1a, 0x2a, 0x0f, 0x08, 0x33, 0x0f, 0xa8, 0xf0, 0x87, 0x2f, 0xc6, 0x21, 0xfa, 0x53,
+	0x0d, 0xae, 0x25, 0x23, 0xd7, 0x17, 0xe8, 0x15, 0xcd, 0x1f, 0x1f, 0xd5, 0xae, 0xe1, 0x4c, 0x69,
+	0x78, 0x48, 0x2b, 0x84, 0x7f, 0x34, 0x10, 0x08, 0xbf, 0x68, 0xff, 0xa8, 0x39, 0x4c, 0x20, 0x1e,
+	0xde, 0x96, 0x97, 0xfe, 0xd1, 0x08, 0xfe, 0xd1, 0x85, 0xbf, 0x49, 0xf8, 0x5e, 0xb6, 0x7f, 0xf4,
+	0x70, 0x74, 0x0b, 0x28, 0xda, 0x91, 0x2f, 0x9d, 0xa2, 0x97, 0x4e, 0xd1, 0x8b, 0x72, 0x8a, 0xfa,
+	0x29, 0xa7, 0x68, 0x9c, 0x64, 0x84, 0xca, 0xf2, 0x4b, 0xb4, 0x61, 0xce, 0xd5, 0xf8, 0x6e, 0xca,
+	0x1f, 0xe7, 0xe0, 0x4a, 0x86, 0x76, 0x40, 0xef, 0xc0, 0xac, 0xc7, 0x1c, 0x97, 0x74, 0x68, 0x34,
+	0xa1, 0xf2, 0x40, 0x13, 0x37, 0x40, 0x9b, 0x29, 0x1a, 0x1e, 0xe0, 0x46, 0x8f, 0x01, 0x88, 0x61,
+	0x50, 0xcf, 0xdb, 0x72, 0x5a, 0x81, 0x5d, 0x76, 0x87, 0xbb, 0x0b, 0xcb, 0x61, 0xe9, 0xb3, 0xa3,
+	0xda, 0x97, 0xb3, 0x12, 0x46, 0x41, 0x7b, 0x98, 0xbc, 0xeb, 0x1d, 0x55, 0xc0, 0x31, 0x48, 0xf4,
+	0xeb, 0x00, 0xf2, 0xf6, 0x77, 0x78, 0x4f, 0xf1, 0x39, 0x51, 0xfd, 0x7a, 0x70, 0xbb, 0xba, 0xfe,
+	0xab, 0x3e, 0xb1, 0x19, 0x5f, 0x15, 0xe2, 0xee, 0xee, 0xc3, 0x10, 0x05, 0xc7, 0x10, 0xf5, 0x7f,
+	0xcc, 0x41, 0x29, 0xb0, 0x17, 0x5f, 0x40, 0xde, 0xa6, 0x93, 0xc8, 0xdb, 0x8c, 0xfe, 0x08, 0x25,
+	0x68, 0xf2, 0xd0, 0x4c, 0x8d, 0x93, 0xca, 0xd4, 0xac, 0x8f, 0x2f, 0xea, 0xe4, 0xdc, 0xcc, 0x33,
+	0x0d, 0x2e, 0x07, 0xac, 0xea, 0x7e, 0xf2, 0x57, 0x61, 0xda, 0xa5, 0xa4, 0xd5, 0x20, 0xcc, 0xd8,
+	0x17, 0xd3, 0x27, 0x6f, 0x27, 0xcf, 0x1d, 0x1f, 0xd5, 0xa6, 0x71, 0x9c, 0x80, 0x93, 0x7c, 0xa8,
+	0x0e, 0xe0, 0xb7, 0xda, 0x8f, 0x1c, 0x57, 0x38, 0x5b, 0xf2, 0x4e, 0xb2, 0x98, 0xc4, 0xdd, 0xd5,
+	0x35, 0x55, 0x8a, 0x63, 0x1c, 0x59, 0x17, 0xa1, 0xf3, 0x63, 0x5e, 0x84, 0x2e, 0x9c, 0xe9, 0x22,
+	0xf4, 0x3f, 0x6b, 0x30, 0x15, 0x75, 0xfe, 0xc2, 0x53, 0x51, 0xed, 0x64, 0x2a, 0x6a, 0x79, 0xec,
+	0xb9, 0x1d, 0x92, 0x7c, 0xfa, 0x41, 0x31, 0xea, 0x96, 0x48, 0x37, 0xed, 0xc1, 0xbc, 0x99, 0x99,
+	0x82, 0x89, 0xa9, 0x8e, 0xf0, 0x5e, 0xd9, 0xc6, 0x50, 0x4e, 0x7c, 0x02, 0x0a, 0xf2, 0xa1, 0x74,
+	0x40, 0x5d, 0x66, 0x1a, 0x34, 0xe8, 0xdf, 0xfa, 0x39, 0x3d, 0x5b, 0x8c, 0xc6, 0xf4, 0xa1, 0x12,
+	0x80, 0x43, 0x51, 0x68, 0x0f, 0x8a, 0xb4, 0xd5, 0xa1, 0xc1, 0x3d, 0xf2, 0xb7, 0xc7, 0xba, 0xc3,
+	0x1f, 0x8d, 0x27, 0xff, 0xf2, 0xb0, 0x84, 0x46, 0x1e, 0x94, 0xad, 0xc0, 0x65, 0x56, 0x06, 0x73,
+	0x63, 0x64, 0x39, 0xa1, 0xf3, 0x1d, 0xdd, 0xeb, 0x0c, 0x8b, 0x70, 0x24, 0x07, 0x75, 0xc3, 0xd7,
+	0x09, 0xc5, 0x73, 0xd2, 0x04, 0x27, 0xbc, 0x7d, 0xf3, 0xa0, 0xfc, 0x84, 0x30, 0xea, 0xf6, 0x88,
+	0xdb, 0x55, 0x06, 0xed, 0xe8, 0x3d, 0x7c, 0x14, 0x20, 0x45, 0x3d, 0x0c, 0x8b, 0x70, 0x24, 0x47,
+	0xff, 0x61, 0x2e, 0x52, 0x3d, 0x2f, 0x3a, 0x17, 0xf8, 0x56, 0x32, 0x17, 0xb8, 0x90, 0xce, 0x05,
+	0xa6, 0x82, 0x11, 0x67, 0xcf, 0x06, 0x12, 0xa8, 0x58, 0xc4, 0x63, 0xbb, 0xfd, 0x16, 0x61, 0x2a,
+	0x2a, 0x57, 0x59, 0xfa, 0xb9, 0xd3, 0x29, 0x93, 0x1d, 0xb3, 0x47, 0x23, 0x7b, 0x75, 0x33, 0x82,
+	0xc1, 0x71, 0x4c, 0xfd, 0x3f, 0x34, 0x98, 0x1b, 0xc8, 0xff, 0xa2, 0x7d, 0x98, 0xb0, 0x85, 0x85,
+	0x3d, 0xf6, 0xa3, 0xba, 0x98, 0xa1, 0x2e, 0x17, 0x8d, 0x2a, 0x50, 0xf8, 0xc8, 0x86, 0x12, 0x7d,
+	0xca, 0xa8, 0x6b, 0x13, 0x4b, 0x1d, 0x8c, 0xe7, 0xf3, 0x80, 0x4f, 0xd8, 0x53, 0x77, 0x15, 0x32,
+	0x0e, 0x65, 0xe8, 0x5f, 0xe4, 0xa0, 0x12, 0xe3, 0x7b, 0x5e, 0xc4, 0x56, 0xdc, 0x2b, 0x94, 0xae,
+	0xe6, 0xae, 0x6b, 0xa9, 0x89, 0x8e, 0xdd, 0x2b, 0x54, 0x24, 0xbc, 0x89, 0xe3, 0x7c, 0x68, 0x09,
+	0xa0, 0x47, 0x3c, 0x46, 0x5d, 0xa1, 0x1b, 0x53, 0xb7, 0xf9, 0xb6, 0x42, 0x0a, 0x8e, 0x71, 0xa1,
+	0x9b, 0x2a, 0xfc, 0x51, 0x48, 0xbe, 0x86, 0x19, 0x12, 0xdb, 0x28, 0x9e, 0x43, 0x6c, 0x03, 0x75,
+	0x60, 0x36, 0x68, 0x75, 0x40, 0x55, 0x1b, 0xf7, 0x94, 0xc0, 0xd2, 0x54, 0x4c, 0x41, 0xe0, 0x01,
+	0x50, 0xfd, 0x6f, 0x34, 0x98, 0x4e, 0xd8, 0xbb, 0xe8, 0xb5, 0xf8, 0xe5, 0x85, 0x58, 0xa4, 0x34,
+	0x71, 0xe9, 0xe0, 0x0d, 0x98, 0x90, 0x03, 0xa4, 0x06, 0x3e, 0xdc, 0x88, 0x72, 0x08, 0xb1, 0xa2,
+	0xf2, 0x2d, 0xa5, 0x42, 0x29, 0xe9, 0x2d, 0xa5, 0x62, 0x2d, 0x38, 0xa0, 0xa3, 0x2f, 0x71, 0x0b,
+	0x5e, 0xb6, 0x4e, 0x8d, 0x74, 0x78, 0x30, 0x04, 0xfd, 0xc0, 0x21, 0x87, 0xfe, 0x36, 0xc8, 0x97,
+	0xb7, 0xe8, 0x3a, 0xe4, 0x7b, 0xa6, 0xad, 0xc2, 0x99, 0x22, 0x68, 0xba, 0x65, 0xda, 0x98, 0x97,
+	0x09, 0x12, 0x79, 0xaa, 0x02, 0x3c, 0x92, 0x44, 0x9e, 0x62, 0x5e, 0x26, 0xde, 0xb7, 0x89, 0xd8,
+	0xf8, 0x23, 0xc8, 0x5b, 0x4e, 0x47, 0x6d, 0xa6, 0xd1, 0x23, 0xb6, 0x9b, 0x4e, 0x47, 0x4a, 0xd8,
+	0x74, 0x3a, 0x98, 0x23, 0x22, 0x03, 0x8a, 0x5d, 0xd2, 0xee, 0x12, 0xb5, 0x77, 0x46, 0xd7, 0xb7,
+	0x61, 0x1c, 0x5f, 0x3d, 0xda, 0xe2, 0x9f, 0x58, 0x62, 0x23, 0x03, 0x26, 0xfc, 0x96, 0xf8, 0x23,
+	0x88, 0x71, 0xff, 0x6b, 0x62, 0x77, 0x55, 0x88, 0x10, 0x8a, 0x40, 0xfe, 0xc6, 0x0a, 0x5a, 0xff,
+	0xa7, 0x1c, 0xa8, 0xbf, 0x9f, 0x40, 0x3e, 0x94, 0x3b, 0xc1, 0x23, 0x29, 0x35, 0x66, 0xef, 0x8e,
+	0x71, 0x2f, 0x37, 0xf1, 0xdc, 0x4a, 0xa6, 0xa6, 0xc2, 0x42, 0x1c, 0x49, 0x42, 0x34, 0x39, 0x96,
+	0xab, 0x63, 0x8e, 0xa5, 0x14, 0x37, 0x38, 0x9a, 0x04, 0x0a, 0xfb, 0x8c, 0xf5, 0xd5, 0x58, 0x8e,
+	0xfe, 0x5e, 0x30, 0xba, 0x1f, 0x27, 0x83, 0xa0, 0xfc, 0x1b, 0x0b, 0x68, 0xbd, 0x07, 0xea, 0xa8,
+	0x42, 0x46, 0xe2, 0x69, 0xa4, 0x0c, 0x6e, 0x2f, 0x9e, 0xee, 0x00, 0x09, 0x9f, 0xf5, 0xc5, 0x5e,
+	0x4c, 0x64, 0xbe, 0x81, 0xd4, 0xff, 0x35, 0x07, 0xf9, 0x9d, 0xcd, 0xa6, 0xbc, 0x00, 0x2c, 0x22,
+	0x15, 0xb4, 0xd9, 0x35, 0xfb, 0x0f, 0xa9, 0x6b, 0xb6, 0xa5, 0x97, 0x5a, 0x8a, 0x5f, 0x00, 0x4e,
+	0x73, 0xe0, 0x8c, 0x5a, 0xe8, 0x03, 0x98, 0x32, 0xc8, 0x0a, 0x75, 0x99, 0x54, 0x3a, 0x67, 0x8b,
+	0xe5, 0x8a, 0x74, 0xe2, 0xca, 0x72, 0x54, 0x1d, 0x27, 0xc0, 0xd0, 0x2e, 0x80, 0x11, 0x41, 0xe7,
+	0xcf, 0x02, 0x2d, 0xdf, 0x82, 0x46, 0xc0, 0x31, 0x20, 0x84, 0xa1, 0xdc, 0xe5, 0xac, 0x02, 0xb5,
+	0x70, 0x16, 0x54, 0xb1, 0x28, 0xef, 0x07, 0x75, 0x71, 0x04, 0xa3, 0xff, 0x58, 0x83, 0xfc, 0xee,
+	0xea, 0x1a, 0x72, 0xa0, 0x1c, 0xde, 0x5a, 0x51, 0x7b, 0xa2, 0x31, 0x7e, 0x18, 0x53, 0x0a, 0x0e,
+	0x3f, 0x71, 0x24, 0x03, 0xed, 0xc3, 0xe4, 0x9e, 0x6f, 0x5a, 0xcc, 0xb4, 0x45, 0x9c, 0x67, 0x1c,
+	0x4f, 0x23, 0x78, 0x5d, 0xa9, 0x92, 0x66, 0x12, 0x15, 0x07, 0xf0, 0xfa, 0x77, 0x40, 0xe9, 0x02,
+	0x6e, 0x41, 0x5e, 0x44, 0x27, 0x43, 0x0b, 0x32, 0xab, 0xa3, 0xfa, 0x3f, 0xe4, 0x60, 0x42, 0xfd,
+	0xed, 0xce, 0xc5, 0xc7, 0x00, 0x68, 0x22, 0x06, 0xb0, 0x32, 0xe6, 0x1f, 0x51, 0x0c, 0x8d, 0x00,
+	0xf4, 0x52, 0x11, 0x80, 0x71, 0xff, 0xf1, 0xe2, 0x39, 0xfe, 0xff, 0x13, 0x98, 0x8a, 0xff, 0x33,
+	0xc6, 0x0b, 0x73, 0xfe, 0xf5, 0x4f, 0x35, 0x80, 0x40, 0xf2, 0x85, 0x7b, 0xde, 0xad, 0xa4, 0xe7,
+	0x7d, 0x67, 0xcc, 0x31, 0x1d, 0xe2, 0x77, 0xff, 0x6d, 0x21, 0xe8, 0x92, 0xf0, 0xba, 0x3f, 0xd6,
+	0xe0, 0x32, 0x49, 0x78, 0xb2, 0xaa, 0x67, 0xe7, 0xe6, 0x18, 0x5f, 0x53, 0xcd, 0x48, 0xfd, 0x07,
+	0x15, 0x4e, 0x89, 0x45, 0xb7, 0x61, 0xaa, 0xaf, 0xfc, 0x19, 0x61, 0xd5, 0xe6, 0x92, 0x77, 0x0b,
+	0xb6, 0x63, 0x34, 0x9c, 0xe0, 0x7c, 0x4e, 0xe4, 0x20, 0x7f, 0x2e, 0x91, 0x83, 0x78, 0xae, 0xad,
+	0x70, 0x62, 0xae, 0xcd, 0x86, 0x72, 0xdb, 0x75, 0x7a, 0xc2, 0x39, 0x57, 0xff, 0x4d, 0x31, 0xa6,
+	0xc3, 0x1f, 0xea, 0x97, 0xb5, 0x00, 0x17, 0x47, 0x22, 0xb8, 0x22, 0x65, 0x8e, 0x94, 0x36, 0x71,
+	0x1e, 0xd2, 0x42, 0xdb, 0x76, 0x47, 0xa2, 0xe2, 0x00, 0x5e, 0xff, 0x97, 0x5c, 0xb0, 0x0f, 0x9b,
+	0xa9, 0xcb, 0xaa, 0xda, 0x90, 0xcb, 0xaa, 0xea, 0xbd, 0x41, 0xdc, 0x3d, 0x7d, 0x03, 0x26, 0x5c,
+	0x4a, 0x3c, 0xc7, 0x56, 0x2f, 0x77, 0xc2, 0x3d, 0x8f, 0x45, 0x29, 0x56, 0xd4, 0xb8, 0x1b, 0x9b,
+	0x7b, 0x8e, 0x1b, 0xfb, 0xa5, 0xd8, 0xdc, 0xe4, 0xc5, 0x9e, 0x0e, 0xb7, 0x59, 0xc6, 0xfc, 0x08,
+	0x0b, 0x5d, 0xa5, 0x89, 0x8a, 0x69, 0x0b, 0x5d, 0xa5, 0x72, 0x42, 0x0e, 0xd4, 0x82, 0x29, 0xee,
+	0xce, 0x0a, 0x2b, 0xbd, 0xb5, 0xcc, 0x46, 0xf0, 0x91, 0xc3, 0x15, 0xbc, 0x19, 0xc3, 0xc1, 0x09,
+	0x54, 0xfd, 0xeb, 0x10, 0x45, 0x1f, 0xd0, 0x22, 0x94, 0xfb, 0xae, 0xd3, 0x27, 0x1d, 0xc2, 0xa8,
+	0xb2, 0x6e, 0xc2, 0x15, 0xb0, 0x1d, 0x10, 0x70, 0xc4, 0xd3, 0xa8, 0x7f, 0xf2, 0xf9, 0xc2, 0xa5,
+	0x4f, 0x3f, 0x5f, 0xb8, 0xf4, 0xd9, 0xe7, 0x0b, 0x97, 0x7e, 0xfb, 0x78, 0x41, 0xfb, 0xe4, 0x78,
+	0x41, 0xfb, 0xf4, 0x78, 0x41, 0xfb, 0xec, 0x78, 0x41, 0xfb, 0xf1, 0xf1, 0x82, 0xf6, 0xbd, 0x7f,
+	0x5b, 0xb8, 0xf4, 0xcd, 0x52, 0x30, 0xcd, 0xff, 0x1b, 0x00, 0x00, 0xff, 0xff, 0x03, 0xe6, 0xe9,
+	0xab, 0xbb, 0x4f, 0x00, 0x00,
 }
 
 func (m *AbstractVertex) Marshal() (dAtA []byte, err error) {
@@ -2388,6 +2389,18 @@ func (m *Edge) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 	_ = i
 	var l int
 	_ = l
+	if m.Limits != nil {
+		{
+			size, err := m.Limits.MarshalToSizedBuffer(dAtA[:i])
+			if err != nil {
+				return 0, err
+			}
+			i -= size
+			i = encodeVarintGenerated(dAtA, i, uint64(size))
+		}
+		i--
+		dAtA[i] = 0x22
+	}
 	if m.Conditions != nil {
 		{
 			size, err := m.Conditions.MarshalToSizedBuffer(dAtA[:i])
@@ -2413,6 +2426,39 @@ func (m *Edge) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 	return len(dAtA) - i, nil
 }
 
+func (m *EdgeLimits) Marshal() (dAtA []byte, err error) {
+	size := m.Size()
+	dAtA = make([]byte, size)
+	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	if err != nil {
+		return nil, err
+	}
+	return dAtA[:n], nil
+}
+
+func (m *EdgeLimits) MarshalTo(dAtA []byte) (int, error) {
+	size := m.Size()
+	return m.MarshalToSizedBuffer(dAtA[:size])
+}
+
+func (m *EdgeLimits) MarshalToSizedBuffer(dAtA []byte) (int, error) {
+	i := len(dAtA)
+	_ = i
+	var l int
+	_ = l
+	if m.BufferUsageLimit != nil {
+		i = encodeVarintGenerated(dAtA, i, uint64(*m.BufferUsageLimit))
+		i--
+		dAtA[i] = 0x10
+	}
+	if m.BufferMaxLength != nil {
+		i = encodeVarintGenerated(dAtA, i, uint64(*m.BufferMaxLength))
+		i--
+		dAtA[i] = 0x8
+	}
+	return len(dAtA) - i, nil
+}
+
 func (m *ForwardConditions) Marshal() (dAtA []byte, err error) {
 	size := m.Size()
 	dAtA = make([]byte, size)
@@ -4744,46 +4790,6 @@ func (m *TLS) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 	return len(dAtA) - i, nil
 }
 
-func (m *ToVertex) Marshal() (dAtA []byte, err error) {
-	size := m.Size()
-	dAtA = make([]byte, size)
-	n, err := m.MarshalToSizedBuffer(dAtA[:size])
-	if err != nil {
-		return nil, err
-	}
-	return dAtA[:n], nil
-}
-
-func (m *ToVertex) MarshalTo(dAtA []byte) (int, error) {
-	size := m.Size()
-	return m.MarshalToSizedBuffer(dAtA[:size])
-}
-
-func (m *ToVertex) MarshalToSizedBuffer(dAtA []byte) (int, error) {
-	i := len(dAtA)
-	_ = i
-	var l int
-	_ = l
-	if m.Conditions != nil {
-		{
-			size, err := m.Conditions.MarshalToSizedBuffer(dAtA[:i])
-			if err != nil {
-				return 0, err
-			}
-			i -= size
-			i = encodeVarintGenerated(dAtA, i, uint64(size))
-		}
-		i--
-		dAtA[i] = 0x12
-	}
-	i -= len(m.Name)
-	copy(dAtA[i:], m.Name)
-	i = encodeVarintGenerated(dAtA, i, uint64(len(m.Name)))
-	i--
-	dAtA[i] = 0xa
-	return len(dAtA) - i, nil
-}
-
 func (m *UDF) Marshal() (dAtA []byte, err error) {
 	size := m.Size()
 	dAtA = make([]byte, size)
@@ -4937,16 +4943,6 @@ func (m *VertexLimits) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 	_ = i
 	var l int
 	_ = l
-	if m.BufferUsageLimit != nil {
-		i = encodeVarintGenerated(dAtA, i, uint64(*m.BufferUsageLimit))
-		i--
-		dAtA[i] = 0x20
-	}
-	if m.BufferMaxLength != nil {
-		i = encodeVarintGenerated(dAtA, i, uint64(*m.BufferMaxLength))
-		i--
-		dAtA[i] = 0x18
-	}
 	if m.UDFWorkers != nil {
 		i = encodeVarintGenerated(dAtA, i, uint64(*m.UDFWorkers))
 		i--
@@ -5027,10 +5023,10 @@ func (m *VertexSpec) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 	_ = i
 	var l int
 	_ = l
-	if len(m.ToVertices) > 0 {
-		for iNdEx := len(m.ToVertices) - 1; iNdEx >= 0; iNdEx-- {
+	if len(m.ToEdges) > 0 {
+		for iNdEx := len(m.ToEdges) - 1; iNdEx >= 0; iNdEx-- {
 			{
-				size, err := m.ToVertices[iNdEx].MarshalToSizedBuffer(dAtA[:i])
+				size, err := m.ToEdges[iNdEx].MarshalToSizedBuffer(dAtA[:i])
 				if err != nil {
 					return 0, err
 				}
@@ -5041,11 +5037,16 @@ func (m *VertexSpec) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 			dAtA[i] = 0x32
 		}
 	}
-	if len(m.FromVertices) > 0 {
-		for iNdEx := len(m.FromVertices) - 1; iNdEx >= 0; iNdEx-- {
-			i -= len(m.FromVertices[iNdEx])
-			copy(dAtA[i:], m.FromVertices[iNdEx])
-			i = encodeVarintGenerated(dAtA, i, uint64(len(m.FromVertices[iNdEx])))
+	if len(m.FromEdges) > 0 {
+		for iNdEx := len(m.FromEdges) - 1; iNdEx >= 0; iNdEx-- {
+			{
+				size, err := m.FromEdges[iNdEx].MarshalToSizedBuffer(dAtA[:i])
+				if err != nil {
+					return 0, err
+				}
+				i -= size
+				i = encodeVarintGenerated(dAtA, i, uint64(size))
+			}
 			i--
 			dAtA[i] = 0x2a
 		}
@@ -5371,6 +5372,25 @@ func (m *Edge) Size() (n int) {
 		l = m.Conditions.Size()
 		n += 1 + l + sovGenerated(uint64(l))
 	}
+	if m.Limits != nil {
+		l = m.Limits.Size()
+		n += 1 + l + sovGenerated(uint64(l))
+	}
+	return n
+}
+
+func (m *EdgeLimits) Size() (n int) {
+	if m == nil {
+		return 0
+	}
+	var l int
+	_ = l
+	if m.BufferMaxLength != nil {
+		n += 1 + sovGenerated(uint64(*m.BufferMaxLength))
+	}
+	if m.BufferUsageLimit != nil {
+		n += 1 + sovGenerated(uint64(*m.BufferUsageLimit))
+	}
 	return n
 }
 
@@ -6237,21 +6257,6 @@ func (m *TLS) Size() (n int) {
 	return n
 }
 
-func (m *ToVertex) Size() (n int) {
-	if m == nil {
-		return 0
-	}
-	var l int
-	_ = l
-	l = len(m.Name)
-	n += 1 + l + sovGenerated(uint64(l))
-	if m.Conditions != nil {
-		l = m.Conditions.Size()
-		n += 1 + l + sovGenerated(uint64(l))
-	}
-	return n
-}
-
 func (m *UDF) Size() (n int) {
 	if m == nil {
 		return 0
@@ -6307,12 +6312,6 @@ func (m *VertexLimits) Size() (n int) {
 	if m.UDFWorkers != nil {
 		n += 1 + sovGenerated(uint64(*m.UDFWorkers))
 	}
-	if m.BufferMaxLength != nil {
-		n += 1 + sovGenerated(uint64(*m.BufferMaxLength))
-	}
-	if m.BufferUsageLimit != nil {
-		n += 1 + sovGenerated(uint64(*m.BufferUsageLimit))
-	}
 	return n
 }
 
@@ -6348,14 +6347,14 @@ func (m *VertexSpec) Size() (n int) {
 	if m.Replicas != nil {
 		n += 1 + sovGenerated(uint64(*m.Replicas))
 	}
-	if len(m.FromVertices) > 0 {
-		for _, s := range m.FromVertices {
-			l = len(s)
+	if len(m.FromEdges) > 0 {
+		for _, e := range m.FromEdges {
+			l = e.Size()
 			n += 1 + l + sovGenerated(uint64(l))
 		}
 	}
-	if len(m.ToVertices) > 0 {
-		for _, e := range m.ToVertices {
+	if len(m.ToEdges) > 0 {
+		for _, e := range m.ToEdges {
 			l = e.Size()
 			n += 1 + l + sovGenerated(uint64(l))
 		}
@@ -6533,6 +6532,18 @@ func (this *Edge) String() string {
 		`From:` + fmt.Sprintf("%v", this.From) + `,`,
 		`To:` + fmt.Sprintf("%v", this.To) + `,`,
 		`Conditions:` + strings.Replace(this.Conditions.String(), "ForwardConditions", "ForwardConditions", 1) + `,`,
+		`Limits:` + strings.Replace(this.Limits.String(), "EdgeLimits", "EdgeLimits", 1) + `,`,
+		`}`,
+	}, "")
+	return s
+}
+func (this *EdgeLimits) String() string {
+	if this == nil {
+		return "nil"
+	}
+	s := strings.Join([]string{`&EdgeLimits{`,
+		`BufferMaxLength:` + valueToStringGenerated(this.BufferMaxLength) + `,`,
+		`BufferUsageLimit:` + valueToStringGenerated(this.BufferUsageLimit) + `,`,
 		`}`,
 	}, "")
 	return s
@@ -7182,17 +7193,6 @@ func (this *TLS) String() string {
 	}, "")
 	return s
 }
-func (this *ToVertex) String() string {
-	if this == nil {
-		return "nil"
-	}
-	s := strings.Join([]string{`&ToVertex{`,
-		`Name:` + fmt.Sprintf("%v", this.Name) + `,`,
-		`Conditions:` + strings.Replace(this.Conditions.String(), "ForwardConditions", "ForwardConditions", 1) + `,`,
-		`}`,
-	}, "")
-	return s
-}
 func (this *UDF) String() string {
 	if this == nil {
 		return "nil"
@@ -7233,8 +7233,6 @@ func (this *VertexLimits) String() string {
 	s := strings.Join([]string{`&VertexLimits{`,
 		`ReadBatchSize:` + valueToStringGenerated(this.ReadBatchSize) + `,`,
 		`UDFWorkers:` + valueToStringGenerated(this.UDFWorkers) + `,`,
-		`BufferMaxLength:` + valueToStringGenerated(this.BufferMaxLength) + `,`,
-		`BufferUsageLimit:` + valueToStringGenerated(this.BufferUsageLimit) + `,`,
 		`}`,
 	}, "")
 	return s
@@ -7259,18 +7257,23 @@ func (this *VertexSpec) String() string {
 	if this == nil {
 		return "nil"
 	}
-	repeatedStringForToVertices := "[]ToVertex{"
-	for _, f := range this.ToVertices {
-		repeatedStringForToVertices += strings.Replace(strings.Replace(f.String(), "ToVertex", "ToVertex", 1), `&`, ``, 1) + ","
+	repeatedStringForFromEdges := "[]Edge{"
+	for _, f := range this.FromEdges {
+		repeatedStringForFromEdges += strings.Replace(strings.Replace(f.String(), "Edge", "Edge", 1), `&`, ``, 1) + ","
 	}
-	repeatedStringForToVertices += "}"
+	repeatedStringForFromEdges += "}"
+	repeatedStringForToEdges := "[]Edge{"
+	for _, f := range this.ToEdges {
+		repeatedStringForToEdges += strings.Replace(strings.Replace(f.String(), "Edge", "Edge", 1), `&`, ``, 1) + ","
+	}
+	repeatedStringForToEdges += "}"
 	s := strings.Join([]string{`&VertexSpec{`,
 		`AbstractVertex:` + strings.Replace(strings.Replace(this.AbstractVertex.String(), "AbstractVertex", "AbstractVertex", 1), `&`, ``, 1) + `,`,
 		`PipelineName:` + fmt.Sprintf("%v", this.PipelineName) + `,`,
 		`InterStepBufferServiceName:` + fmt.Sprintf("%v", this.InterStepBufferServiceName) + `,`,
 		`Replicas:` + valueToStringGenerated(this.Replicas) + `,`,
-		`FromVertices:` + fmt.Sprintf("%v", this.FromVertices) + `,`,
-		`ToVertices:` + repeatedStringForToVertices + `,`,
+		`FromEdges:` + repeatedStringForFromEdges + `,`,
+		`ToEdges:` + repeatedStringForToEdges + `,`,
 		`}`,
 	}, "")
 	return s
@@ -8725,7 +8728,191 @@ func (m *ContainerTemplate) Unmarshal(dAtA []byte) error {
 			iNdEx = postIndex
 		case 4:
 			if wireType != 2 {
-				return fmt.Errorf("proto: wrong wireType = %d for field Env", wireType)
+				return fmt.Errorf("proto: wrong wireType = %d for field Env", wireType)
+			}
+			var msglen int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowGenerated
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				msglen |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			if msglen < 0 {
+				return ErrInvalidLengthGenerated
+			}
+			postIndex := iNdEx + msglen
+			if postIndex < 0 {
+				return ErrInvalidLengthGenerated
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.Env = append(m.Env, v1.EnvVar{})
+			if err := m.Env[len(m.Env)-1].Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
+				return err
+			}
+			iNdEx = postIndex
+		default:
+			iNdEx = preIndex
+			skippy, err := skipGenerated(dAtA[iNdEx:])
+			if err != nil {
+				return err
+			}
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
+				return ErrInvalidLengthGenerated
+			}
+			if (iNdEx + skippy) > l {
+				return io.ErrUnexpectedEOF
+			}
+			iNdEx += skippy
+		}
+	}
+
+	if iNdEx > l {
+		return io.ErrUnexpectedEOF
+	}
+	return nil
+}
+func (m *Edge) Unmarshal(dAtA []byte) error {
+	l := len(dAtA)
+	iNdEx := 0
+	for iNdEx < l {
+		preIndex := iNdEx
+		var wire uint64
+		for shift := uint(0); ; shift += 7 {
+			if shift >= 64 {
+				return ErrIntOverflowGenerated
+			}
+			if iNdEx >= l {
+				return io.ErrUnexpectedEOF
+			}
+			b := dAtA[iNdEx]
+			iNdEx++
+			wire |= uint64(b&0x7F) << shift
+			if b < 0x80 {
+				break
+			}
+		}
+		fieldNum := int32(wire >> 3)
+		wireType := int(wire & 0x7)
+		if wireType == 4 {
+			return fmt.Errorf("proto: Edge: wiretype end group for non-group")
+		}
+		if fieldNum <= 0 {
+			return fmt.Errorf("proto: Edge: illegal tag %d (wire type %d)", fieldNum, wire)
+		}
+		switch fieldNum {
+		case 1:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field From", wireType)
+			}
+			var stringLen uint64
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowGenerated
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				stringLen |= uint64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			intStringLen := int(stringLen)
+			if intStringLen < 0 {
+				return ErrInvalidLengthGenerated
+			}
+			postIndex := iNdEx + intStringLen
+			if postIndex < 0 {
+				return ErrInvalidLengthGenerated
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.From = string(dAtA[iNdEx:postIndex])
+			iNdEx = postIndex
+		case 2:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field To", wireType)
+			}
+			var stringLen uint64
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowGenerated
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				stringLen |= uint64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			intStringLen := int(stringLen)
+			if intStringLen < 0 {
+				return ErrInvalidLengthGenerated
+			}
+			postIndex := iNdEx + intStringLen
+			if postIndex < 0 {
+				return ErrInvalidLengthGenerated
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.To = string(dAtA[iNdEx:postIndex])
+			iNdEx = postIndex
+		case 3:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Conditions", wireType)
+			}
+			var msglen int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowGenerated
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				msglen |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			if msglen < 0 {
+				return ErrInvalidLengthGenerated
+			}
+			postIndex := iNdEx + msglen
+			if postIndex < 0 {
+				return ErrInvalidLengthGenerated
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			if m.Conditions == nil {
+				m.Conditions = &ForwardConditions{}
+			}
+			if err := m.Conditions.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
+				return err
+			}
+			iNdEx = postIndex
+		case 4:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Limits", wireType)
 			}
 			var msglen int
 			for shift := uint(0); ; shift += 7 {
@@ -8752,8 +8939,10 @@ func (m *ContainerTemplate) Unmarshal(dAtA []byte) error {
 			if postIndex > l {
 				return io.ErrUnexpectedEOF
 			}
-			m.Env = append(m.Env, v1.EnvVar{})
-			if err := m.Env[len(m.Env)-1].Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
+			if m.Limits == nil {
+				m.Limits = &EdgeLimits{}
+			}
+			if err := m.Limits.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
 				return err
 			}
 			iNdEx = postIndex
@@ -8778,7 +8967,7 @@ func (m *ContainerTemplate) Unmarshal(dAtA []byte) error {
 	}
 	return nil
 }
-func (m *Edge) Unmarshal(dAtA []byte) error {
+func (m *EdgeLimits) Unmarshal(dAtA []byte) error {
 	l := len(dAtA)
 	iNdEx := 0
 	for iNdEx < l {
@@ -8801,17 +8990,17 @@ func (m *Edge) Unmarshal(dAtA []byte) error {
 		fieldNum := int32(wire >> 3)
 		wireType := int(wire & 0x7)
 		if wireType == 4 {
-			return fmt.Errorf("proto: Edge: wiretype end group for non-group")
+			return fmt.Errorf("proto: EdgeLimits: wiretype end group for non-group")
 		}
 		if fieldNum <= 0 {
-			return fmt.Errorf("proto: Edge: illegal tag %d (wire type %d)", fieldNum, wire)
+			return fmt.Errorf("proto: EdgeLimits: illegal tag %d (wire type %d)", fieldNum, wire)
 		}
 		switch fieldNum {
 		case 1:
-			if wireType != 2 {
-				return fmt.Errorf("proto: wrong wireType = %d for field From", wireType)
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field BufferMaxLength", wireType)
 			}
-			var stringLen uint64
+			var v uint64
 			for shift := uint(0); ; shift += 7 {
 				if shift >= 64 {
 					return ErrIntOverflowGenerated
@@ -8821,61 +9010,17 @@ func (m *Edge) Unmarshal(dAtA []byte) error {
 				}
 				b := dAtA[iNdEx]
 				iNdEx++
-				stringLen |= uint64(b&0x7F) << shift
+				v |= uint64(b&0x7F) << shift
 				if b < 0x80 {
 					break
 				}
 			}
-			intStringLen := int(stringLen)
-			if intStringLen < 0 {
-				return ErrInvalidLengthGenerated
-			}
-			postIndex := iNdEx + intStringLen
-			if postIndex < 0 {
-				return ErrInvalidLengthGenerated
-			}
-			if postIndex > l {
-				return io.ErrUnexpectedEOF
-			}
-			m.From = string(dAtA[iNdEx:postIndex])
-			iNdEx = postIndex
+			m.BufferMaxLength = &v
 		case 2:
-			if wireType != 2 {
-				return fmt.Errorf("proto: wrong wireType = %d for field To", wireType)
-			}
-			var stringLen uint64
-			for shift := uint(0); ; shift += 7 {
-				if shift >= 64 {
-					return ErrIntOverflowGenerated
-				}
-				if iNdEx >= l {
-					return io.ErrUnexpectedEOF
-				}
-				b := dAtA[iNdEx]
-				iNdEx++
-				stringLen |= uint64(b&0x7F) << shift
-				if b < 0x80 {
-					break
-				}
-			}
-			intStringLen := int(stringLen)
-			if intStringLen < 0 {
-				return ErrInvalidLengthGenerated
-			}
-			postIndex := iNdEx + intStringLen
-			if postIndex < 0 {
-				return ErrInvalidLengthGenerated
-			}
-			if postIndex > l {
-				return io.ErrUnexpectedEOF
-			}
-			m.To = string(dAtA[iNdEx:postIndex])
-			iNdEx = postIndex
-		case 3:
-			if wireType != 2 {
-				return fmt.Errorf("proto: wrong wireType = %d for field Conditions", wireType)
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field BufferUsageLimit", wireType)
 			}
-			var msglen int
+			var v uint32
 			for shift := uint(0); ; shift += 7 {
 				if shift >= 64 {
 					return ErrIntOverflowGenerated
@@ -8885,28 +9030,12 @@ func (m *Edge) Unmarshal(dAtA []byte) error {
 				}
 				b := dAtA[iNdEx]
 				iNdEx++
-				msglen |= int(b&0x7F) << shift
+				v |= uint32(b&0x7F) << shift
 				if b < 0x80 {
 					break
 				}
 			}
-			if msglen < 0 {
-				return ErrInvalidLengthGenerated
-			}
-			postIndex := iNdEx + msglen
-			if postIndex < 0 {
-				return ErrInvalidLengthGenerated
-			}
-			if postIndex > l {
-				return io.ErrUnexpectedEOF
-			}
-			if m.Conditions == nil {
-				m.Conditions = &ForwardConditions{}
-			}
-			if err := m.Conditions.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
-				return err
-			}
-			iNdEx = postIndex
+			m.BufferUsageLimit = &v
 		default:
 			iNdEx = preIndex
 			skippy, err := skipGenerated(dAtA[iNdEx:])
@@ -16775,124 +16904,6 @@ func (m *TLS) Unmarshal(dAtA []byte) error {
 	}
 	return nil
 }
-func (m *ToVertex) Unmarshal(dAtA []byte) error {
-	l := len(dAtA)
-	iNdEx := 0
-	for iNdEx < l {
-		preIndex := iNdEx
-		var wire uint64
-		for shift := uint(0); ; shift += 7 {
-			if shift >= 64 {
-				return ErrIntOverflowGenerated
-			}
-			if iNdEx >= l {
-				return io.ErrUnexpectedEOF
-			}
-			b := dAtA[iNdEx]
-			iNdEx++
-			wire |= uint64(b&0x7F) << shift
-			if b < 0x80 {
-				break
-			}
-		}
-		fieldNum := int32(wire >> 3)
-		wireType := int(wire & 0x7)
-		if wireType == 4 {
-			return fmt.Errorf("proto: ToVertex: wiretype end group for non-group")
-		}
-		if fieldNum <= 0 {
-			return fmt.Errorf("proto: ToVertex: illegal tag %d (wire type %d)", fieldNum, wire)
-		}
-		switch fieldNum {
-		case 1:
-			if wireType != 2 {
-				return fmt.Errorf("proto: wrong wireType = %d for field Name", wireType)
-			}
-			var stringLen uint64
-			for shift := uint(0); ; shift += 7 {
-				if shift >= 64 {
-					return ErrIntOverflowGenerated
-				}
-				if iNdEx >= l {
-					return io.ErrUnexpectedEOF
-				}
-				b := dAtA[iNdEx]
-				iNdEx++
-				stringLen |= uint64(b&0x7F) << shift
-				if b < 0x80 {
-					break
-				}
-			}
-			intStringLen := int(stringLen)
-			if intStringLen < 0 {
-				return ErrInvalidLengthGenerated
-			}
-			postIndex := iNdEx + intStringLen
-			if postIndex < 0 {
-				return ErrInvalidLengthGenerated
-			}
-			if postIndex > l {
-				return io.ErrUnexpectedEOF
-			}
-			m.Name = string(dAtA[iNdEx:postIndex])
-			iNdEx = postIndex
-		case 2:
-			if wireType != 2 {
-				return fmt.Errorf("proto: wrong wireType = %d for field Conditions", wireType)
-			}
-			var msglen int
-			for shift := uint(0); ; shift += 7 {
-				if shift >= 64 {
-					return ErrIntOverflowGenerated
-				}
-				if iNdEx >= l {
-					return io.ErrUnexpectedEOF
-				}
-				b := dAtA[iNdEx]
-				iNdEx++
-				msglen |= int(b&0x7F) << shift
-				if b < 0x80 {
-					break
-				}
-			}
-			if msglen < 0 {
-				return ErrInvalidLengthGenerated
-			}
-			postIndex := iNdEx + msglen
-			if postIndex < 0 {
-				return ErrInvalidLengthGenerated
-			}
-			if postIndex > l {
-				return io.ErrUnexpectedEOF
-			}
-			if m.Conditions == nil {
-				m.Conditions = &ForwardConditions{}
-			}
-			if err := m.Conditions.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
-				return err
-			}
-			iNdEx = postIndex
-		default:
-			iNdEx = preIndex
-			skippy, err := skipGenerated(dAtA[iNdEx:])
-			if err != nil {
-				return err
-			}
-			if (skippy < 0) || (iNdEx+skippy) < 0 {
-				return ErrInvalidLengthGenerated
-			}
-			if (iNdEx + skippy) > l {
-				return io.ErrUnexpectedEOF
-			}
-			iNdEx += skippy
-		}
-	}
-
-	if iNdEx > l {
-		return io.ErrUnexpectedEOF
-	}
-	return nil
-}
 func (m *UDF) Unmarshal(dAtA []byte) error {
 	l := len(dAtA)
 	iNdEx := 0
@@ -17316,46 +17327,6 @@ func (m *VertexLimits) Unmarshal(dAtA []byte) error {
 				}
 			}
 			m.UDFWorkers = &v
-		case 3:
-			if wireType != 0 {
-				return fmt.Errorf("proto: wrong wireType = %d for field BufferMaxLength", wireType)
-			}
-			var v uint64
-			for shift := uint(0); ; shift += 7 {
-				if shift >= 64 {
-					return ErrIntOverflowGenerated
-				}
-				if iNdEx >= l {
-					return io.ErrUnexpectedEOF
-				}
-				b := dAtA[iNdEx]
-				iNdEx++
-				v |= uint64(b&0x7F) << shift
-				if b < 0x80 {
-					break
-				}
-			}
-			m.BufferMaxLength = &v
-		case 4:
-			if wireType != 0 {
-				return fmt.Errorf("proto: wrong wireType = %d for field BufferUsageLimit", wireType)
-			}
-			var v uint32
-			for shift := uint(0); ; shift += 7 {
-				if shift >= 64 {
-					return ErrIntOverflowGenerated
-				}
-				if iNdEx >= l {
-					return io.ErrUnexpectedEOF
-				}
-				b := dAtA[iNdEx]
-				iNdEx++
-				v |= uint32(b&0x7F) << shift
-				if b < 0x80 {
-					break
-				}
-			}
-			m.BufferUsageLimit = &v
 		default:
 			iNdEx = preIndex
 			skippy, err := skipGenerated(dAtA[iNdEx:])
@@ -17642,9 +17613,9 @@ func (m *VertexSpec) Unmarshal(dAtA []byte) error {
 			m.Replicas = &v
 		case 5:
 			if wireType != 2 {
-				return fmt.Errorf("proto: wrong wireType = %d for field FromVertices", wireType)
+				return fmt.Errorf("proto: wrong wireType = %d for field FromEdges", wireType)
 			}
-			var stringLen uint64
+			var msglen int
 			for shift := uint(0); ; shift += 7 {
 				if shift >= 64 {
 					return ErrIntOverflowGenerated
@@ -17654,27 +17625,29 @@ func (m *VertexSpec) Unmarshal(dAtA []byte) error {
 				}
 				b := dAtA[iNdEx]
 				iNdEx++
-				stringLen |= uint64(b&0x7F) << shift
+				msglen |= int(b&0x7F) << shift
 				if b < 0x80 {
 					break
 				}
 			}
-			intStringLen := int(stringLen)
-			if intStringLen < 0 {
+			if msglen < 0 {
 				return ErrInvalidLengthGenerated
 			}
-			postIndex := iNdEx + intStringLen
+			postIndex := iNdEx + msglen
 			if postIndex < 0 {
 				return ErrInvalidLengthGenerated
 			}
 			if postIndex > l {
 				return io.ErrUnexpectedEOF
 			}
-			m.FromVertices = append(m.FromVertices, string(dAtA[iNdEx:postIndex]))
+			m.FromEdges = append(m.FromEdges, Edge{})
+			if err := m.FromEdges[len(m.FromEdges)-1].Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
+				return err
+			}
 			iNdEx = postIndex
 		case 6:
 			if wireType != 2 {
-				return fmt.Errorf("proto: wrong wireType = %d for field ToVertices", wireType)
+				return fmt.Errorf("proto: wrong wireType = %d for field ToEdges", wireType)
 			}
 			var msglen int
 			for shift := uint(0); ; shift += 7 {
@@ -17701,8 +17674,8 @@ func (m *VertexSpec) Unmarshal(dAtA []byte) error {
 			if postIndex > l {
 				return io.ErrUnexpectedEOF
 			}
-			m.ToVertices = append(m.ToVertices, ToVertex{})
-			if err := m.ToVertices[len(m.ToVertices)-1].Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
+			m.ToEdges = append(m.ToEdges, Edge{})
+			if err := m.ToEdges[len(m.ToEdges)-1].Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
 				return err
 			}
 			iNdEx = postIndex
diff --git a/pkg/apis/numaflow/v1alpha1/generated.proto b/pkg/apis/numaflow/v1alpha1/generated.proto
index f68c5101..f6df41b2 100644
--- a/pkg/apis/numaflow/v1alpha1/generated.proto
+++ b/pkg/apis/numaflow/v1alpha1/generated.proto
@@ -170,6 +170,24 @@ message Edge {
   // Conditional forwarding, only allowed when "From" is a Sink or UDF
   // +optional
   optional ForwardConditions conditions = 3;
+
+  // Limits define the limitations such as buffer read batch size for the edge, will override pipeline level settings
+  // +optional
+  optional EdgeLimits limits = 4;
+}
+
+message EdgeLimits {
+  // BufferMaxLength is used to define the max length of a buffer.
+  // It overrides the settings from pipeline limits.
+  // Only meaningful for UDF and Source vertice as only they do buffer write.
+  // +optional
+  optional uint64 bufferMaxLength = 1;
+
+  // BufferUsageLimit is used to define the pencentage of the buffer usage limit, a valid value should be less than 100, for example, 85.
+  // It overrides the settings from pipeline limits.
+  // Only meaningful for UDF and Source vertice as only they do buffer write.
+  // +optional
+  optional uint32 bufferUsageLimit = 2;
 }
 
 message ForwardConditions {
@@ -839,13 +857,6 @@ message TLS {
   optional k8s.io.api.core.v1.SecretKeySelector keySecret = 4;
 }
 
-message ToVertex {
-  optional string name = 1;
-
-  // +optional
-  optional ForwardConditions conditions = 2;
-}
-
 message UDF {
   // +optional
   optional Container container = 1;
@@ -888,18 +899,6 @@ message VertexLimits {
   // It overrides the setting in pipeline limits.
   // +optional
   optional uint32 udfWorkers = 2;
-
-  // BufferMaxLength is used to define the max length of a buffer.
-  // It overrides the settings from pipeline limits.
-  // Only meaningful for UDF and Source vertice as only they do buffer write.
-  // +optional
-  optional uint64 bufferMaxLength = 3;
-
-  // BufferUsageLimit is used to define the pencentage of the buffer usage limit, a valid value should be less than 100, for example, 85.
-  // It overrides the settings from pipeline limits.
-  // Only meaningful for UDF and Source vertice as only they do buffer write.
-  // +optional
-  optional uint32 bufferUsageLimit = 4;
 }
 
 // +kubebuilder:object:root=true
@@ -923,10 +922,10 @@ message VertexSpec {
   optional int32 replicas = 4;
 
   // +optional
-  repeated string fromVertices = 5;
+  repeated Edge fromEdges = 5;
 
   // +optional
-  repeated ToVertex toVertices = 6;
+  repeated Edge toEdges = 6;
 }
 
 message VertexStatus {
diff --git a/pkg/apis/numaflow/v1alpha1/pipeline_types.go b/pkg/apis/numaflow/v1alpha1/pipeline_types.go
index 2b4a8e03..87f332ef 100644
--- a/pkg/apis/numaflow/v1alpha1/pipeline_types.go
+++ b/pkg/apis/numaflow/v1alpha1/pipeline_types.go
@@ -77,23 +77,14 @@ func (p Pipeline) GetVertex(vertexName string) *AbstractVertex {
 	return nil
 }
 
-// FindVerticesWithEdgeBuffer is used to locate the vertices who write and read from the buffer.
-func (p Pipeline) FindVerticesWithEdgeBuffer(buffer string) (from, to *AbstractVertex) {
+// FindEdgeWithBuffer is used to locate the edge of the buffer.
+func (p Pipeline) FindEdgeWithBuffer(buffer string) *Edge {
 	for _, e := range p.Spec.Edges {
 		if buffer == GenerateEdgeBufferName(p.Namespace, p.Name, e.From, e.To) {
-			for _, v := range p.Spec.Vertices {
-				if v.Name == e.From {
-					from = v.DeepCopy()
-				} else if v.Name == e.To {
-					to = v.DeepCopy()
-				}
-				if from != nil && to != nil {
-					return from, to
-				}
-			}
+			return &e
 		}
 	}
-	return from, to
+	return nil
 }
 
 func (p Pipeline) GetToEdges(vertexName string) []Edge {
@@ -321,18 +312,6 @@ type PipelineLimits struct {
 	BufferUsageLimit *uint32 `json:"bufferUsageLimit,omitempty" protobuf:"varint,4,opt,name=bufferUsageLimit"`
 }
 
-type Edge struct {
-	From string `json:"from" protobuf:"bytes,1,opt,name=from"`
-	To   string `json:"to" protobuf:"bytes,2,opt,name=to"`
-	// Conditional forwarding, only allowed when "From" is a Sink or UDF
-	// +optional
-	Conditions *ForwardConditions `json:"conditions" protobuf:"bytes,3,opt,name=conditions"`
-}
-
-type ForwardConditions struct {
-	KeyIn []string `json:"keyIn" protobuf:"bytes,1,rep,name=keyIn"`
-}
-
 type PipelineStatus struct {
 	Status      `json:",inline" protobuf:"bytes,1,opt,name=status"`
 	Phase       PipelinePhase `json:"phase,omitempty" protobuf:"bytes,2,opt,name=phase,casttype=PipelinePhase"`
diff --git a/pkg/apis/numaflow/v1alpha1/vertex_types.go b/pkg/apis/numaflow/v1alpha1/vertex_types.go
index b08a4532..7ef27cb4 100644
--- a/pkg/apis/numaflow/v1alpha1/vertex_types.go
+++ b/pkg/apis/numaflow/v1alpha1/vertex_types.go
@@ -259,8 +259,8 @@ func (v Vertex) GetFromBuffers() []Buffer {
 	if v.IsASource() {
 		r = append(r, Buffer{GenerateSourceBufferName(v.Namespace, v.Spec.PipelineName, v.Spec.Name), SourceBuffer})
 	} else {
-		for _, vt := range v.Spec.FromVertices {
-			r = append(r, Buffer{GenerateEdgeBufferName(v.Namespace, v.Spec.PipelineName, vt, v.Spec.Name), EdgeBuffer})
+		for _, vt := range v.Spec.FromEdges {
+			r = append(r, Buffer{GenerateEdgeBufferName(v.Namespace, v.Spec.PipelineName, vt.From, v.Spec.Name), EdgeBuffer})
 		}
 	}
 	return r
@@ -271,8 +271,8 @@ func (v Vertex) GetToBuffers() []Buffer {
 	if v.IsASink() {
 		r = append(r, Buffer{GenerateSinkBufferName(v.Namespace, v.Spec.PipelineName, v.Spec.Name), SinkBuffer})
 	} else {
-		for _, vt := range v.Spec.ToVertices {
-			r = append(r, Buffer{GenerateEdgeBufferName(v.Namespace, v.Spec.PipelineName, v.Spec.Name, vt.Name), EdgeBuffer})
+		for _, vt := range v.Spec.ToEdges {
+			r = append(r, Buffer{GenerateEdgeBufferName(v.Namespace, v.Spec.PipelineName, v.Spec.Name, vt.To), EdgeBuffer})
 		}
 	}
 	return r
@@ -287,15 +287,9 @@ type VertexSpec struct {
 	// +optional
 	Replicas *int32 `json:"replicas,omitempty" protobuf:"varint,4,opt,name=replicas"`
 	// +optional
-	FromVertices []string `json:"fromVertices,omitempty" protobuf:"bytes,5,rep,name=fromVertices"`
+	FromEdges []Edge `json:"fromEdges,omitempty" protobuf:"bytes,5,rep,name=fromEdges"`
 	// +optional
-	ToVertices []ToVertex `json:"toVertices,omitempty" protobuf:"bytes,6,rep,name=toVertices"`
-}
-
-type ToVertex struct {
-	Name string `json:"name" protobuf:"bytes,1,opt,name=name"`
-	// +optional
-	Conditions *ForwardConditions `json:"conditions" protobuf:"bytes,2,opt,name=conditions"`
+	ToEdges []Edge `json:"toEdges,omitempty" protobuf:"bytes,6,rep,name=toEdges"`
 }
 
 func (vs VertexSpec) GetReplicas() int {
@@ -390,16 +384,6 @@ type VertexLimits struct {
 	// It overrides the setting in pipeline limits.
 	// +optional
 	UDFWorkers *uint32 `json:"udfWorkers,omitempty" protobuf:"varint,2,opt,name=udfWorkers"`
-	// BufferMaxLength is used to define the max length of a buffer.
-	// It overrides the settings from pipeline limits.
-	// Only meaningful for UDF and Source vertice as only they do buffer write.
-	// +optional
-	BufferMaxLength *uint64 `json:"bufferMaxLength,omitempty" protobuf:"varint,3,opt,name=bufferMaxLength"`
-	// BufferUsageLimit is used to define the pencentage of the buffer usage limit, a valid value should be less than 100, for example, 85.
-	// It overrides the settings from pipeline limits.
-	// Only meaningful for UDF and Source vertice as only they do buffer write.
-	// +optional
-	BufferUsageLimit *uint32 `json:"bufferUsageLimit,omitempty" protobuf:"varint,4,opt,name=bufferUsageLimit"`
 }
 
 func (v VertexSpec) getType() containerSupplier {
diff --git a/pkg/apis/numaflow/v1alpha1/zz_generated.deepcopy.go b/pkg/apis/numaflow/v1alpha1/zz_generated.deepcopy.go
index e08fe322..6c2677e6 100644
--- a/pkg/apis/numaflow/v1alpha1/zz_generated.deepcopy.go
+++ b/pkg/apis/numaflow/v1alpha1/zz_generated.deepcopy.go
@@ -256,6 +256,11 @@ func (in *Edge) DeepCopyInto(out *Edge) {
 		*out = new(ForwardConditions)
 		(*in).DeepCopyInto(*out)
 	}
+	if in.Limits != nil {
+		in, out := &in.Limits, &out.Limits
+		*out = new(EdgeLimits)
+		(*in).DeepCopyInto(*out)
+	}
 	return
 }
 
@@ -269,6 +274,32 @@ func (in *Edge) DeepCopy() *Edge {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *EdgeLimits) DeepCopyInto(out *EdgeLimits) {
+	*out = *in
+	if in.BufferMaxLength != nil {
+		in, out := &in.BufferMaxLength, &out.BufferMaxLength
+		*out = new(uint64)
+		**out = **in
+	}
+	if in.BufferUsageLimit != nil {
+		in, out := &in.BufferUsageLimit, &out.BufferUsageLimit
+		*out = new(uint32)
+		**out = **in
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new EdgeLimits.
+func (in *EdgeLimits) DeepCopy() *EdgeLimits {
+	if in == nil {
+		return nil
+	}
+	out := new(EdgeLimits)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *ForwardConditions) DeepCopyInto(out *ForwardConditions) {
 	*out = *in
@@ -1347,27 +1378,6 @@ func (in *TLS) DeepCopy() *TLS {
 	return out
 }
 
-// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
-func (in *ToVertex) DeepCopyInto(out *ToVertex) {
-	*out = *in
-	if in.Conditions != nil {
-		in, out := &in.Conditions, &out.Conditions
-		*out = new(ForwardConditions)
-		(*in).DeepCopyInto(*out)
-	}
-	return
-}
-
-// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ToVertex.
-func (in *ToVertex) DeepCopy() *ToVertex {
-	if in == nil {
-		return nil
-	}
-	out := new(ToVertex)
-	in.DeepCopyInto(out)
-	return out
-}
-
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *UDF) DeepCopyInto(out *UDF) {
 	*out = *in
@@ -1452,16 +1462,6 @@ func (in *VertexLimits) DeepCopyInto(out *VertexLimits) {
 		*out = new(uint32)
 		**out = **in
 	}
-	if in.BufferMaxLength != nil {
-		in, out := &in.BufferMaxLength, &out.BufferMaxLength
-		*out = new(uint64)
-		**out = **in
-	}
-	if in.BufferUsageLimit != nil {
-		in, out := &in.BufferUsageLimit, &out.BufferUsageLimit
-		*out = new(uint32)
-		**out = **in
-	}
 	return
 }
 
@@ -1517,14 +1517,16 @@ func (in *VertexSpec) DeepCopyInto(out *VertexSpec) {
 		*out = new(int32)
 		**out = **in
 	}
-	if in.FromVertices != nil {
-		in, out := &in.FromVertices, &out.FromVertices
-		*out = make([]string, len(*in))
-		copy(*out, *in)
+	if in.FromEdges != nil {
+		in, out := &in.FromEdges, &out.FromEdges
+		*out = make([]Edge, len(*in))
+		for i := range *in {
+			(*in)[i].DeepCopyInto(&(*out)[i])
+		}
 	}
-	if in.ToVertices != nil {
-		in, out := &in.ToVertices, &out.ToVertices
-		*out = make([]ToVertex, len(*in))
+	if in.ToEdges != nil {
+		in, out := &in.ToEdges, &out.ToEdges
+		*out = make([]Edge, len(*in))
 		for i := range *in {
 			(*in)[i].DeepCopyInto(&(*out)[i])
 		}
diff --git a/pkg/daemon/server/service/isbs_query_service.go b/pkg/daemon/server/service/isbs_query_service.go
index 6e4ca390..59306a43 100644
--- a/pkg/daemon/server/service/isbs_query_service.go
+++ b/pkg/daemon/server/service/isbs_query_service.go
@@ -29,28 +29,22 @@ func (is *isbSvcQueryService) ListBuffers(ctx context.Context, req *daemon.ListB
 	resp := new(daemon.ListBuffersResponse)
 
 	buffers := []*daemon.BufferInfo{}
-	for _, buffer := range is.pipeline.GetAllBuffers() {
-		if buffer.Type != v1alpha1.EdgeBuffer {
-			continue
-		}
-		bufferInfo, err := is.client.GetBufferInfo(ctx, buffer)
+	for _, edge := range is.pipeline.Spec.Edges {
+		buffer := v1alpha1.GenerateEdgeBufferName(is.pipeline.Namespace, is.pipeline.Name, edge.From, edge.To)
+		bufferInfo, err := is.client.GetBufferInfo(ctx, v1alpha1.Buffer{Name: buffer, Type: v1alpha1.EdgeBuffer})
 		if err != nil {
 			return nil, fmt.Errorf("failed to get information of buffer %q", buffer)
 		}
 		log.Debugf("Buffer %s has bufferInfo %+v", buffer, bufferInfo)
-		vFrom, vTo := is.pipeline.FindVerticesWithEdgeBuffer(buffer.Name)
-		if vFrom == nil || vTo == nil {
-			return nil, fmt.Errorf("buffer %q not found from the pipeline", buffer)
-		}
-		bufferLength, bufferUsageLimit := getVertexLimits(is.pipeline, vFrom)
+		bufferLength, bufferUsageLimit := getBufferLimits(is.pipeline, edge)
 		usage := float64(bufferInfo.TotalMessages) / float64(bufferLength)
 		if x := (float64(bufferInfo.PendingCount) + float64(bufferInfo.AckPendingCount)) / float64(bufferLength); x < usage {
 			usage = x
 		}
 		b := &daemon.BufferInfo{
 			Pipeline:         &is.pipeline.Name,
-			FromVertex:       &vFrom.Name,
-			ToVertex:         &vTo.Name,
+			FromVertex:       &edge.From,
+			ToVertex:         &edge.To,
 			BufferName:       pointer.String(fmt.Sprintf("%v", buffer)),
 			PendingCount:     &bufferInfo.PendingCount,
 			AckPendingCount:  &bufferInfo.AckPendingCount,
@@ -72,19 +66,19 @@ func (is *isbSvcQueryService) GetBuffer(ctx context.Context, req *daemon.GetBuff
 	if err != nil {
 		return nil, fmt.Errorf("failed to get information of buffer %q", *req.Buffer)
 	}
-	vFrom, vTo := is.pipeline.FindVerticesWithEdgeBuffer(*req.Buffer)
-	if vFrom == nil || vTo == nil {
+	edge := is.pipeline.FindEdgeWithBuffer(*req.Buffer)
+	if edge == nil {
 		return nil, fmt.Errorf("unexpected error, buffer %q not found from the pipeline", *req.Buffer)
 	}
-	bufferLength, bufferUsageLimit := getVertexLimits(is.pipeline, vFrom)
+	bufferLength, bufferUsageLimit := getBufferLimits(is.pipeline, *edge)
 	usage := float64(bufferInfo.TotalMessages) / float64(bufferLength)
 	if x := (float64(bufferInfo.PendingCount) + float64(bufferInfo.AckPendingCount)) / float64(bufferLength); x < usage {
 		usage = x
 	}
 	b := &daemon.BufferInfo{
 		Pipeline:         &is.pipeline.Name,
-		FromVertex:       &vFrom.Name,
-		ToVertex:         &vTo.Name,
+		FromVertex:       &edge.From,
+		ToVertex:         &edge.To,
 		BufferName:       req.Buffer,
 		PendingCount:     &bufferInfo.PendingCount,
 		AckPendingCount:  &bufferInfo.AckPendingCount,
@@ -99,7 +93,7 @@ func (is *isbSvcQueryService) GetBuffer(ctx context.Context, req *daemon.GetBuff
 	return resp, nil
 }
 
-func getVertexLimits(pl *v1alpha1.Pipeline, v *v1alpha1.AbstractVertex) (bufferLength int64, bufferUsageLimit float64) {
+func getBufferLimits(pl *v1alpha1.Pipeline, edge v1alpha1.Edge) (bufferLength int64, bufferUsageLimit float64) {
 	bufferLength = int64(v1alpha1.DefaultBufferLength)
 	bufferUsageLimit = v1alpha1.DefaultBufferUsageLimit
 	if x := pl.Spec.Limits; x != nil {
@@ -110,7 +104,7 @@ func getVertexLimits(pl *v1alpha1.Pipeline, v *v1alpha1.AbstractVertex) (bufferL
 			bufferUsageLimit = float64(*x.BufferUsageLimit) / 100
 		}
 	}
-	if x := v.Limits; x != nil {
+	if x := edge.Limits; x != nil {
 		if x.BufferMaxLength != nil {
 			bufferLength = int64(*x.BufferMaxLength)
 		}
diff --git a/pkg/sources/source.go b/pkg/sources/source.go
index 3c6d6c16..3676e896 100644
--- a/pkg/sources/source.go
+++ b/pkg/sources/source.go
@@ -32,38 +32,35 @@ func (u *SourceProcessor) Start(ctx context.Context) error {
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 	var writers []isb.BufferWriter
-	toBuffers := u.Vertex.GetToBuffers()
 	switch u.ISBSvcType {
 	case dfv1.ISBSvcTypeRedis:
-		writeOpts := []redisisb.Option{}
-		if x := u.Vertex.Spec.Limits; x != nil {
-			if x.BufferMaxLength != nil {
+		for _, e := range u.Vertex.Spec.ToEdges {
+			writeOpts := []redisisb.Option{}
+			if x := e.Limits; x != nil && x.BufferMaxLength != nil {
 				writeOpts = append(writeOpts, redisisb.WithMaxLength(int64(*x.BufferMaxLength)))
 			}
-			if x.BufferUsageLimit != nil {
+			if x := e.Limits; x != nil && x.BufferUsageLimit != nil {
 				writeOpts = append(writeOpts, redisisb.WithBufferUsageLimit(float64(*x.BufferUsageLimit)/100))
 			}
-		}
-		redisClient := clients.NewInClusterRedisClient()
-		for _, b := range toBuffers {
-			group := b.Name + "-group"
-			writer := redisisb.NewBufferWrite(ctx, redisClient, b.Name, group, writeOpts...)
+			buffer := dfv1.GenerateEdgeBufferName(u.Vertex.Namespace, u.Vertex.Spec.PipelineName, e.From, e.To)
+			group := buffer + "-group"
+			redisClient := clients.NewInClusterRedisClient()
+			writer := redisisb.NewBufferWrite(ctx, redisClient, buffer, group, writeOpts...)
 			writers = append(writers, writer)
 		}
 	case dfv1.ISBSvcTypeJetStream:
-		writeOpts := []jetstreamisb.WriteOption{}
-		if x := u.Vertex.Spec.Limits; x != nil {
-			if x.BufferMaxLength != nil {
+		for _, e := range u.Vertex.Spec.ToEdges {
+			writeOpts := []jetstreamisb.WriteOption{}
+			if x := e.Limits; x != nil && x.BufferMaxLength != nil {
 				writeOpts = append(writeOpts, jetstreamisb.WithMaxLength(int64(*x.BufferMaxLength)))
 			}
-			if x.BufferUsageLimit != nil {
+			if x := e.Limits; x != nil && x.BufferUsageLimit != nil {
 				writeOpts = append(writeOpts, jetstreamisb.WithBufferUsageLimit(float64(*x.BufferUsageLimit)/100))
 			}
-		}
-		for _, buffer := range toBuffers {
-			streamName := fmt.Sprintf("%s-%s", u.Vertex.Spec.PipelineName, buffer.Name)
+			buffer := dfv1.GenerateEdgeBufferName(u.Vertex.Namespace, u.Vertex.Spec.PipelineName, e.From, e.To)
+			streamName := fmt.Sprintf("%s-%s", u.Vertex.Spec.PipelineName, buffer)
 			jetStreamClient := clients.NewInClusterJetStreamClient()
-			writer, err := jetstreamisb.NewJetStreamBufferWriter(ctx, jetStreamClient, buffer.Name, streamName, streamName, writeOpts...)
+			writer, err := jetstreamisb.NewJetStreamBufferWriter(ctx, jetStreamClient, buffer, streamName, streamName, writeOpts...)
 			if err != nil {
 				return err
 			}
@@ -77,8 +74,7 @@ func (u *SourceProcessor) Start(ctx context.Context) error {
 	if err != nil {
 		return fmt.Errorf("failed to find a sourcer, error: %w", err)
 	}
-
-	log.Infow("Start processing source messages", zap.String("isbs", string(u.ISBSvcType)), zap.Any("to", toBuffers))
+	log.Infow("Start processing source messages", zap.String("isbs", string(u.ISBSvcType)), zap.Any("to", u.Vertex.GetToBuffers()))
 	stopped := sourcer.Start()
 	wg := &sync.WaitGroup{}
 	wg.Add(1)
diff --git a/pkg/udf/udf.go b/pkg/udf/udf.go
index e0010f71..d5092208 100644
--- a/pkg/udf/udf.go
+++ b/pkg/udf/udf.go
@@ -44,42 +44,39 @@ func (u *UDFProcessor) Start(ctx context.Context) error {
 		fromGroup := fromBufferName + "-group"
 		consumer := fmt.Sprintf("%s-%v", u.Vertex.Name, u.Replica)
 		reader = redisisb.NewBufferRead(ctx, redisClient, fromBufferName, fromGroup, consumer)
-		writeOpts := []redisisb.Option{}
-		if x := u.Vertex.Spec.Limits; x != nil {
-			if x.BufferMaxLength != nil {
+		for _, e := range u.Vertex.Spec.ToEdges {
+			writeOpts := []redisisb.Option{}
+			if x := e.Limits; x != nil && x.BufferMaxLength != nil {
 				writeOpts = append(writeOpts, redisisb.WithMaxLength(int64(*x.BufferMaxLength)))
 			}
-			if x.BufferUsageLimit != nil {
+			if x := e.Limits; x != nil && x.BufferUsageLimit != nil {
 				writeOpts = append(writeOpts, redisisb.WithBufferUsageLimit(float64(*x.BufferUsageLimit)/100))
 			}
-		}
-		for _, buffer := range toBuffers {
-			writer := redisisb.NewBufferWrite(ctx, redisClient, buffer.Name, buffer.Name+"-group", writeOpts...)
-			writers[buffer.Name] = writer
+			buffer := dfv1.GenerateEdgeBufferName(u.Vertex.Namespace, u.Vertex.Spec.PipelineName, e.From, e.To)
+			writer := redisisb.NewBufferWrite(ctx, redisClient, buffer, buffer+"-group", writeOpts...)
+			writers[buffer] = writer
 		}
 	case dfv1.ISBSvcTypeJetStream:
 		fromStreamName := fmt.Sprintf("%s-%s", u.Vertex.Spec.PipelineName, fromBufferName)
-		jetStreamClient := clients.NewInClusterJetStreamClient()
-		reader, err = jetstreamisb.NewJetStreamBufferReader(ctx, jetStreamClient, fromBufferName, fromStreamName, fromStreamName)
+		reader, err = jetstreamisb.NewJetStreamBufferReader(ctx, clients.NewInClusterJetStreamClient(), fromBufferName, fromStreamName, fromStreamName)
 		if err != nil {
 			return err
 		}
-		writeOpts := []jetstreamisb.WriteOption{}
-		if x := u.Vertex.Spec.Limits; x != nil {
-			if x.BufferMaxLength != nil {
+		for _, e := range u.Vertex.Spec.ToEdges {
+			writeOpts := []jetstreamisb.WriteOption{}
+			if x := e.Limits; x != nil && x.BufferMaxLength != nil {
 				writeOpts = append(writeOpts, jetstreamisb.WithMaxLength(int64(*x.BufferMaxLength)))
 			}
-			if x.BufferUsageLimit != nil {
+			if x := e.Limits; x != nil && x.BufferUsageLimit != nil {
 				writeOpts = append(writeOpts, jetstreamisb.WithBufferUsageLimit(float64(*x.BufferUsageLimit)/100))
 			}
-		}
-		for _, buffer := range toBuffers {
-			streamName := fmt.Sprintf("%s-%s", u.Vertex.Spec.PipelineName, buffer.Name)
-			writer, err := jetstreamisb.NewJetStreamBufferWriter(ctx, jetStreamClient, buffer.Name, streamName, streamName, writeOpts...)
+			buffer := dfv1.GenerateEdgeBufferName(u.Vertex.Namespace, u.Vertex.Spec.PipelineName, e.From, e.To)
+			streamName := fmt.Sprintf("%s-%s", u.Vertex.Spec.PipelineName, buffer)
+			writer, err := jetstreamisb.NewJetStreamBufferWriter(ctx, clients.NewInClusterJetStreamClient(), buffer, streamName, streamName, writeOpts...)
 			if err != nil {
 				return err
 			}
-			writers[buffer.Name] = writer
+			writers[buffer] = writer
 		}
 	default:
 		return fmt.Errorf("unrecognized isbs type %q", u.ISBSvcType)
@@ -92,11 +89,11 @@ func (u *UDFProcessor) Start(ctx context.Context) error {
 			result = append(result, _key)
 			return result, nil
 		}
-		for _, to := range u.Vertex.Spec.ToVertices {
+		for _, to := range u.Vertex.Spec.ToEdges {
 			// If returned key is not "ALL" or "DROP", and there's no conditions defined in the edge,
 			// treat it as "ALL"?
 			if to.Conditions == nil || len(to.Conditions.KeyIn) == 0 || sharedutil.StringSliceContains(to.Conditions.KeyIn, _key) {
-				result = append(result, dfv1.GenerateEdgeBufferName(u.Vertex.Namespace, u.Vertex.Spec.PipelineName, u.Vertex.Spec.Name, to.Name))
+				result = append(result, dfv1.GenerateEdgeBufferName(u.Vertex.Namespace, u.Vertex.Spec.PipelineName, to.From, to.To))
 			}
 		}
 		return result, nil
diff --git a/pkg/watermark/progress/utils.go b/pkg/watermark/progress/utils.go
index a831260e..ed2576c2 100644
--- a/pkg/watermark/progress/utils.go
+++ b/pkg/watermark/progress/utils.go
@@ -33,9 +33,9 @@ func GetHeartbeatBucket(js nats.JetStreamContext, publishKeyspace string) (nats.
 
 // GetFetchKeyspace gets the fetch keyspace name from the vertex.
 func GetFetchKeyspace(v *dfv1.Vertex) string {
-	if len(v.Spec.FromVertices) > 0 {
+	if len(v.Spec.FromEdges) > 0 {
 		// from vertices is 0 because we do not support diamond DAG
-		return fmt.Sprintf("%s-%s-%s", v.Namespace, v.Spec.PipelineName, v.Spec.FromVertices[0])
+		return fmt.Sprintf("%s-%s-%s", v.Namespace, v.Spec.PipelineName, v.Spec.FromEdges[0].From)
 	} else {
 		// sources will not have FromVertices
 		return fmt.Sprintf("%s-%s-%s-source", v.Namespace, v.Spec.PipelineName, v.Name)
diff --git a/server/cmd/main.go b/server/cmd/start.go
similarity index 97%
rename from server/cmd/main.go
rename to server/cmd/start.go
index 87277923..071c7d16 100644
--- a/server/cmd/main.go
+++ b/server/cmd/start.go
@@ -1,4 +1,4 @@
-package main
+package cmd
 
 import (
 	"crypto/tls"
@@ -18,7 +18,7 @@ var (
 	}
 )
 
-func main() {
+func Start() {
 	router := gin.New()
 	router.Use(gin.Logger())
 	router.RedirectTrailingSlash = true


Test patch:
diff --git a/cmd/commands/commands_test.go b/cmd/commands/commands_test.go
index edd59c5f..fbb2c75c 100644
--- a/cmd/commands/commands_test.go
+++ b/cmd/commands/commands_test.go
@@ -136,8 +136,8 @@ func generateEncodedVertexSpecs() string {
 		},
 		Spec: dfv1.VertexSpec{
 			Replicas:     &replicas,
-			FromVertices: []string{"input"},
-			ToVertices:   []dfv1.ToVertex{{Name: "output"}},
+			FromEdges:    []dfv1.Edge{{From: "input", To: "name"}},
+			ToEdges:      []dfv1.Edge{{From: "name", To: "output"}},
 			PipelineName: "test-pl",
 			AbstractVertex: dfv1.AbstractVertex{
 				Name: "name",
diff --git a/controllers/pipeline/controller_test.go b/controllers/pipeline/controller_test.go
index 4dfcc39d..45396e5f 100644
--- a/controllers/pipeline/controller_test.go
+++ b/controllers/pipeline/controller_test.go
@@ -132,28 +132,56 @@ func Test_buildVertices(t *testing.T) {
 	assert.True(t, existing)
 }
 
-func Test_copyLimits(t *testing.T) {
+func Test_copyVertexLimits(t *testing.T) {
 	pl := testPipeline.DeepCopy()
 	v := pl.Spec.Vertices[0].DeepCopy()
-	copyLimits(pl, v)
+	copyVertexLimits(pl, v)
 	assert.Nil(t, v.Limits)
 	one := uint64(1)
 	int32One := uint32(1)
 	pl.Spec.Limits = &dfv1.PipelineLimits{ReadBatchSize: &one, UDFWorkers: &int32One}
-	copyLimits(pl, v)
+	copyVertexLimits(pl, v)
 	assert.NotNil(t, v.Limits)
 	assert.Equal(t, one, *v.Limits.ReadBatchSize)
 	assert.Nil(t, v.Limits.UDFWorkers)
 	two := uint64(2)
 	v.Limits.ReadBatchSize = &two
-	copyLimits(pl, v)
+	copyVertexLimits(pl, v)
 	assert.Equal(t, two, *v.Limits.ReadBatchSize)
 	assert.Nil(t, v.Limits.UDFWorkers)
 
 	v1 := pl.Spec.Vertices[1].DeepCopy()
-	copyLimits(pl, v1)
+	copyVertexLimits(pl, v1)
 	assert.Equal(t, int32One, *v1.Limits.UDFWorkers)
+}
+
+func Test_copyEdgeLimits(t *testing.T) {
+	pl := testPipeline.DeepCopy()
+	edges := []dfv1.Edge{{From: "in", To: "out"}}
+	result := copyEdgeLimits(pl, edges)
+	for _, e := range result {
+		assert.Nil(t, e.Limits)
+	}
+	onethouand := uint64(1000)
+	eighty := uint32(80)
+	pl.Spec.Limits = &dfv1.PipelineLimits{BufferMaxLength: &onethouand, BufferUsageLimit: &eighty}
+	result = copyEdgeLimits(pl, edges)
+	for _, e := range result {
+		assert.NotNil(t, e.Limits)
+		assert.NotNil(t, e.Limits.BufferMaxLength)
+		assert.NotNil(t, e.Limits.BufferUsageLimit)
+	}
 
+	twothouand := uint64(2000)
+	edges = []dfv1.Edge{{From: "in", To: "out", Limits: &dfv1.EdgeLimits{BufferMaxLength: &twothouand}}}
+	result = copyEdgeLimits(pl, edges)
+	for _, e := range result {
+		assert.NotNil(t, e.Limits)
+		assert.NotNil(t, e.Limits.BufferMaxLength)
+		assert.Equal(t, twothouand, *e.Limits.BufferMaxLength)
+		assert.NotNil(t, e.Limits.BufferUsageLimit)
+		assert.Equal(t, eighty, *e.Limits.BufferUsageLimit)
+	}
 }
 
 func Test_buildISBBatchJob(t *testing.T) {
diff --git a/controllers/vertex/controller_test.go b/controllers/vertex/controller_test.go
index 9bf49025..c2ca698d 100644
--- a/controllers/vertex/controller_test.go
+++ b/controllers/vertex/controller_test.go
@@ -72,8 +72,8 @@ var (
 		},
 		Spec: dfv1.VertexSpec{
 			Replicas:     &testReplicas,
-			FromVertices: []string{"input"},
-			ToVertices:   []dfv1.ToVertex{{Name: "output"}},
+			FromEdges:    []dfv1.Edge{{From: "input", To: testVertexSpecName}},
+			ToEdges:      []dfv1.Edge{{From: testVertexSpecName, To: "output"}},
 			PipelineName: testPipelineName,
 			AbstractVertex: dfv1.AbstractVertex{
 				Name: testVertexSpecName,
@@ -88,7 +88,7 @@ var (
 		},
 		Spec: dfv1.VertexSpec{
 			Replicas:     &testReplicas,
-			ToVertices:   []dfv1.ToVertex{{Name: "p1"}},
+			ToEdges:      []dfv1.Edge{{From: "input", To: "p1"}},
 			PipelineName: testPipelineName,
 			AbstractVertex: dfv1.AbstractVertex{
 				Name:   "input",
@@ -204,8 +204,8 @@ func Test_BuildPodSpec(t *testing.T) {
 		testObj.Name = "test-pl-output"
 		testObj.Spec.Name = "output"
 		testObj.Spec.Sink = &dfv1.Sink{}
-		testObj.Spec.FromVertices = []string{"p1"}
-		testObj.Spec.ToVertices = []dfv1.ToVertex{}
+		testObj.Spec.FromEdges = []dfv1.Edge{{From: "p1", To: "output"}}
+		testObj.Spec.ToEdges = []dfv1.Edge{}
 		spec, err := r.buildPodSpec(testObj, testPipeline, fakeIsbSvcConfig)
 		assert.NoError(t, err)
 		assert.Equal(t, 1, len(spec.InitContainers))
@@ -248,8 +248,8 @@ func Test_BuildPodSpec(t *testing.T) {
 				},
 			},
 		}
-		testObj.Spec.FromVertices = []string{"p1"}
-		testObj.Spec.ToVertices = []dfv1.ToVertex{}
+		testObj.Spec.FromEdges = []dfv1.Edge{{From: "p1", To: "output"}}
+		testObj.Spec.ToEdges = []dfv1.Edge{}
 		spec, err := r.buildPodSpec(testObj, testPipeline, fakeIsbSvcConfig)
 		assert.NoError(t, err)
 		assert.Equal(t, 1, len(spec.InitContainers))
diff --git a/pkg/apis/numaflow/v1alpha1/pipeline_types_test.go b/pkg/apis/numaflow/v1alpha1/pipeline_types_test.go
index 18aba603..3aa04830 100644
--- a/pkg/apis/numaflow/v1alpha1/pipeline_types_test.go
+++ b/pkg/apis/numaflow/v1alpha1/pipeline_types_test.go
@@ -67,15 +67,13 @@ func Test_GetVertex(t *testing.T) {
 	assert.NotNil(t, v)
 }
 
-func Test_FindVertexWithEdgeBuffer(t *testing.T) {
-	vFrom, vTo := testPipeline.FindVerticesWithEdgeBuffer("nonono")
-	assert.Nil(t, vFrom)
-	assert.Nil(t, vTo)
-	vFrom, vTo = testPipeline.FindVerticesWithEdgeBuffer(testPipeline.Namespace + "-" + testPipeline.Name + "-input-p1")
-	assert.NotNil(t, vFrom)
-	assert.Equal(t, "input", vFrom.Name)
-	assert.NotNil(t, vTo)
-	assert.Equal(t, "p1", vTo.Name)
+func Test_FindEdgeWithBuffer(t *testing.T) {
+	e := testPipeline.FindEdgeWithBuffer("nonono")
+	assert.Nil(t, e)
+	e = testPipeline.FindEdgeWithBuffer(testPipeline.Namespace + "-" + testPipeline.Name + "-input-p1")
+	assert.NotNil(t, e)
+	assert.Equal(t, "input", e.From)
+	assert.Equal(t, "p1", e.To)
 }
 
 func TestGetDaemonServiceName(t *testing.T) {
diff --git a/pkg/apis/numaflow/v1alpha1/vertex_type_test.go b/pkg/apis/numaflow/v1alpha1/vertex_types_test.go
similarity index 97%
rename from pkg/apis/numaflow/v1alpha1/vertex_type_test.go
rename to pkg/apis/numaflow/v1alpha1/vertex_types_test.go
index 1678c677..c386484c 100644
--- a/pkg/apis/numaflow/v1alpha1/vertex_type_test.go
+++ b/pkg/apis/numaflow/v1alpha1/vertex_types_test.go
@@ -33,7 +33,7 @@ var (
 				Name:   testVertexSpecName,
 				Source: &Source{},
 			},
-			ToVertices: []ToVertex{{Name: "output"}},
+			ToEdges: []Edge{{From: testVertexSpecName, To: "output"}},
 		},
 	}
 
@@ -49,7 +49,7 @@ var (
 				Name: testVertexSpecName,
 				Sink: &Sink{},
 			},
-			FromVertices: []string{"input"},
+			FromEdges: []Edge{{From: "input", To: testVertexSpecName}},
 		},
 	}
 
@@ -64,8 +64,8 @@ var (
 			AbstractVertex: AbstractVertex{
 				Name: testVertexSpecName,
 			},
-			FromVertices: []string{"input"},
-			ToVertices:   []ToVertex{{Name: "output"}},
+			FromEdges: []Edge{{From: "input", To: testVertexSpecName}},
+			ToEdges:   []Edge{{From: testVertexSpecName, To: "output"}},
 		},
 	}
 )
diff --git a/test/e2e/functional_test.go b/test/e2e/functional_test.go
index c3fce49b..f08cf46b 100644
--- a/test/e2e/functional_test.go
+++ b/test/e2e/functional_test.go
@@ -11,12 +11,6 @@ import (
 	"github.com/stretchr/testify/suite"
 )
 
-const (
-	LogSourceVertexStarted = "Start processing source messages"
-	LogSinkVertexStarted   = "Start processing sink messages"
-	LogUDFVertexStarted    = "Start processing udf messages"
-)
-
 type FunctionalSuite struct {
 	E2ESuite
 }
@@ -95,6 +89,47 @@ func (s *FunctionalSuite) TestFiltering() {
 	w.Expect().VertexPodLogNotContains("out", "expect[0-2]", PodLogCheckOptionWithTimeout(2*time.Second))
 }
 
+func (s *FunctionalSuite) TestConditionalForwarding() {
+	w := s.Given().Pipeline("@testdata/even-odd.yaml").
+		When().
+		CreatePipelineAndWait()
+
+	defer w.DeletePipelineAndWait()
+
+	w.Expect().
+		VertexPodsRunning().
+		VertexPodLogContains("in", LogSourceVertexStarted).
+		VertexPodLogContains("even-or-odd", LogUDFVertexStarted, PodLogCheckOptionWithContainer("main")).
+		VertexPodLogContains("even-sink", LogSinkVertexStarted).
+		VertexPodLogContains("odd-sink", LogSinkVertexStarted).
+		VertexPodLogContains("number-sink", LogSinkVertexStarted)
+
+	defer w.VertexPodPortForward("in", 8443, dfv1.VertexHTTPSPort).
+		TerminateAllPodPortForwards()
+
+	HTTPExpect(s.T(), "https://localhost:8443").POST("/vertices/in").WithBytes([]byte("888888")).
+		Expect().
+		Status(204)
+	HTTPExpect(s.T(), "https://localhost:8443").POST("/vertices/in").WithBytes([]byte("888889")).
+		Expect().
+		Status(204)
+	HTTPExpect(s.T(), "https://localhost:8443").POST("/vertices/in").WithBytes([]byte("not an interger")).
+		Expect().
+		Status(204)
+
+	w.Expect().VertexPodLogContains("even-sink", "888888")
+	w.Expect().VertexPodLogNotContains("even-sink", "888889", PodLogCheckOptionWithTimeout(2*time.Second))
+	w.Expect().VertexPodLogNotContains("even-sink", "not an interger", PodLogCheckOptionWithTimeout(2*time.Second))
+
+	w.Expect().VertexPodLogContains("odd-sink", "888889")
+	w.Expect().VertexPodLogNotContains("odd-sink", "888888", PodLogCheckOptionWithTimeout(2*time.Second))
+	w.Expect().VertexPodLogNotContains("odd-sink", "not an interger", PodLogCheckOptionWithTimeout(2*time.Second))
+
+	w.Expect().VertexPodLogContains("number-sink", "888888")
+	w.Expect().VertexPodLogContains("number-sink", "888889")
+	w.Expect().VertexPodLogNotContains("number-sink", "not an interger", PodLogCheckOptionWithTimeout(2*time.Second))
+}
+
 func TestFunctionalSuite(t *testing.T) {
 	suite.Run(t, new(FunctionalSuite))
 }
diff --git a/test/e2e/testdata/even-odd.yaml b/test/e2e/testdata/even-odd.yaml
new file mode 100644
index 00000000..9c81b448
--- /dev/null
+++ b/test/e2e/testdata/even-odd.yaml
@@ -0,0 +1,41 @@
+apiVersion: numaflow.numaproj.io/v1alpha1
+kind: Pipeline
+metadata:
+  name: even-odd
+spec:
+  vertices:
+    - name: in
+      source:
+        http: {}
+    - name: even-or-odd
+      udf:
+        container:
+          image: quay.io/numaio/go-even-odd-example
+    - name: even-sink
+      sink:
+        log: {}
+    - name: odd-sink
+      sink:
+        log: {}
+    - name: number-sink
+      sink:
+        log: {}
+  edges:
+    - from: in
+      to: even-or-odd
+    - from: even-or-odd
+      to: even-sink
+      conditions:
+        keyIn:
+          - even
+    - from: even-or-odd
+      to: odd-sink
+      conditions:
+        keyIn:
+          - odd
+    - from: even-or-odd
+      to: number-sink
+      conditions:
+        keyIn:
+          - even
+          - odd
diff --git a/test/fixtures/e2e_suite.go b/test/fixtures/e2e_suite.go
index 532fffcd..648ef009 100644
--- a/test/fixtures/e2e_suite.go
+++ b/test/fixtures/e2e_suite.go
@@ -33,6 +33,10 @@ const (
 	LabelValue     = "true"
 	ISBSvcName     = "numaflow-e2e"
 	defaultTimeout = 60 * time.Second
+
+	LogSourceVertexStarted = "Start processing source messages"
+	LogSinkVertexStarted   = "Start processing sink messages"
+	LogUDFVertexStarted    = "Start processing udf messages"
 )
 
 var (
diff --git a/test/fixtures/util.go b/test/fixtures/util.go
index 3d864b83..c50abac1 100644
--- a/test/fixtures/util.go
+++ b/test/fixtures/util.go
@@ -248,20 +248,16 @@ func PodsLogNotContains(ctx context.Context, kubeClient kubernetes.Interface, na
 	for _, p := range podList.Items {
 		go func(podName string) {
 			fmt.Printf("Watching POD: %s\n", podName)
-			contains, err := podLogContains(cctx, kubeClient, namespace, podName, o.container, regex)
-			if err != nil {
+			if err := podLogContains(cctx, kubeClient, namespace, podName, o.container, regex, resultChan, errChan); err != nil {
 				errChan <- err
 				return
 			}
-			if contains {
-				resultChan <- true
-			}
 		}(p.Name)
 	}
 	for {
 		select {
 		case <-cctx.Done():
-			return true
+			return true // Consider timeout as not containing
 		case result := <-resultChan:
 			if result {
 				return false
@@ -295,30 +291,28 @@ func PodsLogContains(ctx context.Context, kubeClient kubernetes.Interface, names
 	for _, p := range podList.Items {
 		go func(podName string) {
 			fmt.Printf("Watching POD: %s\n", podName)
-			var contains bool
-			var err error
-			if o.count == -1 {
-				contains, err = podLogContains(cctx, kubeClient, namespace, podName, o.container, regex)
-			} else {
-				contains, err = podLogContainsCount(cctx, kubeClient, namespace, podName, o.container, regex, o.count)
-			}
-			if err != nil {
+			if err := podLogContains(cctx, kubeClient, namespace, podName, o.container, regex, resultChan, errChan); err != nil {
 				errChan <- err
 				return
 			}
-			if contains {
-				resultChan <- true
-			}
 		}(p.Name)
 	}
 
+	matchTimes := 0
 	for {
 		select {
 		case <-cctx.Done():
-			return false
+			return false // Consider timeout as false
 		case result := <-resultChan:
 			if result {
-				return true
+				if o.count < 0 {
+					return true
+				} else {
+					matchTimes++
+					if matchTimes >= o.count {
+						return true
+					}
+				}
 			}
 		case err := <-errChan:
 			fmt.Printf("error: %v", err)
@@ -326,79 +320,33 @@ func PodsLogContains(ctx context.Context, kubeClient kubernetes.Interface, names
 	}
 }
 
-func podLogContains(ctx context.Context, client kubernetes.Interface, namespace, podName, containerName, regex string) (bool, error) {
+func podLogContains(ctx context.Context, client kubernetes.Interface, namespace, podName, containerName, regex string, result chan bool, errs chan error) error {
 	stream, err := client.CoreV1().Pods(namespace).GetLogs(podName, &corev1.PodLogOptions{Follow: true, Container: containerName}).Stream(ctx)
 	if err != nil {
-		return false, err
+		return err
 	}
 	defer func() { _ = stream.Close() }()
 
 	exp, err := regexp.Compile(regex)
 	if err != nil {
-		return false, err
+		return err
 	}
 
 	s := bufio.NewScanner(stream)
 	for {
 		select {
 		case <-ctx.Done():
-			return false, nil
+			return nil
 		default:
 			if !s.Scan() {
-				return false, s.Err()
+				return s.Err()
 			}
 			data := s.Bytes()
 			fmt.Println(string(data))
 			if exp.Match(data) {
-				return true, nil
-			}
-		}
-	}
-}
-
-func podLogContainsCount(ctx context.Context, client kubernetes.Interface, namespace, podName, containerName, regex string, count int) (bool, error) {
-	stream, err := client.CoreV1().Pods(namespace).GetLogs(podName, &corev1.PodLogOptions{Follow: true, Container: containerName}).Stream(ctx)
-	if err != nil {
-		return false, err
-	}
-	defer func() { _ = stream.Close() }()
-
-	exp, err := regexp.Compile(regex)
-	if err != nil {
-		return false, err
-	}
-
-	instancesChan := make(chan struct{})
-
-	// scan the log looking for matches
-	go func(ctx context.Context, instancesChan chan<- struct{}) {
-		s := bufio.NewScanner(stream)
-		for {
-			select {
-			case <-ctx.Done():
-				return
-			default:
-				if !s.Scan() {
-					return
-				}
-				data := s.Bytes()
-				fmt.Println(string(data))
-				if exp.Match(data) {
-					instancesChan <- struct{}{}
-				}
+				result <- true
 			}
 		}
-	}(ctx, instancesChan)
-
-	actualCount := 0
-	for {
-		select {
-		case <-instancesChan:
-			actualCount++
-		case <-ctx.Done():
-			fmt.Printf("time:%v, count:%d,actualCount:%d\n", time.Now().Unix(), count, actualCount)
-			return count == actualCount, nil
-		}
 	}
 }
 
diff --git a/test/http-e2e/http_test.go b/test/http-e2e/http_test.go
new file mode 100644
index 00000000..59c14f45
--- /dev/null
+++ b/test/http-e2e/http_test.go
@@ -0,0 +1,94 @@
+//go:build test
+
+package http_e2e
+
+import (
+	"fmt"
+	"testing"
+
+	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
+	. "github.com/numaproj/numaflow/test/fixtures"
+	"github.com/stretchr/testify/suite"
+)
+
+//go:generate kubectl apply -f testdata/http-auth-fake-secret.yaml -n numaflow-system
+
+type HTTPSuite struct {
+	E2ESuite
+}
+
+func (s *HTTPSuite) TestHTTPSourcePipeline() {
+	w := s.Given().Pipeline("@testdata/http-source.yaml").
+		When().
+		CreatePipelineAndWait()
+
+	defer w.DeletePipelineAndWait()
+
+	w.Expect().
+		VertexPodsRunning().
+		VertexPodLogContains("in", LogSourceVertexStarted).
+		VertexPodLogContains("out", LogSinkVertexStarted)
+
+	defer w.VertexPodPortForward("in", 8443, dfv1.VertexHTTPSPort).
+		TerminateAllPodPortForwards()
+
+	// Check Service
+	cmd := fmt.Sprintf("kubectl -n %s get svc -lnumaflow.numaproj.io/pipeline-name=%s,numaflow.numaproj.io/vertex-name=%s | grep -v CLUSTER-IP | grep -v headless", Namespace, "http-source", "in")
+	w.Exec("sh", []string{"-c", cmd}, OutputRegexp("http-source-in"))
+
+	HTTPExpect(s.T(), "https://localhost:8443").GET("/health").Expect().Status(204)
+
+	HTTPExpect(s.T(), "https://localhost:8443").POST("/vertices/in").WithBytes([]byte("no-id")).
+		Expect().
+		Status(204)
+	HTTPExpect(s.T(), "https://localhost:8443").POST("/vertices/in").WithBytes([]byte("no-id")).
+		Expect().
+		Status(204)
+	// No x-numaflow-id, expect 2 outputs
+	w.Expect().VertexPodLogContains("out", "no-id", PodLogCheckOptionWithCount(2))
+
+	HTTPExpect(s.T(), "https://localhost:8443").POST("/vertices/in").WithHeader("x-numaflow-id", "101").WithBytes([]byte("with-id")).
+		Expect().
+		Status(204)
+	HTTPExpect(s.T(), "https://localhost:8443").POST("/vertices/in").WithHeader("x-numaflow-id", "101").WithBytes([]byte("with-id")).
+		Expect().
+		Status(204)
+	// With same x-numaflow-id, expect 1 output
+	w.Expect().VertexPodLogContains("out", "with-id", PodLogCheckOptionWithCount(1))
+
+	HTTPExpect(s.T(), "https://localhost:8443").POST("/vertices/in").WithHeader("x-numaflow-id", "102").WithBytes([]byte("with-id")).
+		Expect().
+		Status(204)
+	// With a new x-numaflow-id, expect 2 outputs
+	w.Expect().VertexPodLogContains("out", "with-id", PodLogCheckOptionWithCount(2))
+}
+
+func (s *HTTPSuite) TestHTTPSourceAuthPipeline() {
+	w := s.Given().Pipeline("@testdata/http-source-with-auth.yaml").
+		When().
+		CreatePipelineAndWait()
+
+	defer w.DeletePipelineAndWait()
+
+	w.Expect().
+		VertexPodsRunning().
+		VertexPodLogContains("in", LogSourceVertexStarted).
+		VertexPodLogContains("out", LogSinkVertexStarted)
+
+	defer w.VertexPodPortForward("in", 8443, dfv1.VertexHTTPSPort).
+		TerminateAllPodPortForwards()
+
+	HTTPExpect(s.T(), "https://localhost:8443").POST("/vertices/in").WithBytes([]byte("no-auth")).
+		Expect().
+		Status(403)
+
+	HTTPExpect(s.T(), "https://localhost:8443").POST("/vertices/in").WithHeader("Authorization", "Bearer faketoken").WithBytes([]byte("with-auth")).
+		Expect().
+		Status(204)
+
+	w.Expect().VertexPodLogContains("out", "with-auth")
+}
+
+func TestHTTPSuite(t *testing.T) {
+	suite.Run(t, new(HTTPSuite))
+}
diff --git a/test/http-e2e/testdata/http-auth-fake-secret.yaml b/test/http-e2e/testdata/http-auth-fake-secret.yaml
new file mode 100644
index 00000000..7d375986
--- /dev/null
+++ b/test/http-e2e/testdata/http-auth-fake-secret.yaml
@@ -0,0 +1,8 @@
+apiVersion: v1
+kind: Secret
+metadata:
+  name: http-source-fake-token
+  labels:
+    numaflow-e2e: "true"
+stringData:
+  fake-token: "faketoken"
diff --git a/test/http-e2e/testdata/http-source-with-auth.yaml b/test/http-e2e/testdata/http-source-with-auth.yaml
new file mode 100644
index 00000000..a0825395
--- /dev/null
+++ b/test/http-e2e/testdata/http-source-with-auth.yaml
@@ -0,0 +1,19 @@
+apiVersion: numaflow.numaproj.io/v1alpha1
+kind: Pipeline
+metadata:
+  name: http-auth-source
+spec:
+  vertices:
+    - name: in
+      source:
+        http:
+          auth:
+            token:
+              name: http-source-fake-token
+              key: fake-token
+    - name: out
+      sink:
+        log: {}
+  edges:
+    - from: in
+      to: out
diff --git a/test/http-e2e/testdata/http-source.yaml b/test/http-e2e/testdata/http-source.yaml
new file mode 100644
index 00000000..4139a9fa
--- /dev/null
+++ b/test/http-e2e/testdata/http-source.yaml
@@ -0,0 +1,16 @@
+apiVersion: numaflow.numaproj.io/v1alpha1
+kind: Pipeline
+metadata:
+  name: http-source
+spec:
+  vertices:
+    - name: in
+      source:
+        http:
+          service: true
+    - name: out
+      sink:
+        log: {}
+  edges:
+    - from: in
+      to: out