1. 首先定义路由，路由提供了流式的调用方式，包括输入端、处理函数和输出端。不同Endpoint其路由处理是一致的

router := endpoint.NewRouter().From("/api/v1/msg/").Process(func(exchange *endpoint.Exchange) bool {
//处理逻辑
return true
}).To("chain:default")

提示

不同Endpoint类型，输入端From代表的含义会有不同，但最终会根据From值路由到该路由器：

• http/websocket endpoint：代表路径路由，根据From值创建指定的http服务。例如：From("/api/v1/msg/")表示创建/api/v1/msg/ http服务。
• mqtt/kafka endpoint：代表订阅的主题，根据From值订阅相关主题。例如：From("/api/v1/msg/")表示订阅/api/v1/msg/主题。
• schedule endpoint：代表cron表达式，根据From值创建相关定时任务。例如：From("*/1 * * * * *")表示每隔1秒触发该路由器。
• tpc/udp endpoint：代表正则表达式，根据From值把满足条件的消息转发到该路由。例如：From("^{.*")表示满足{开头的数据。

2. 然后创建Endpoint服务，创建接口也是一致的：

//例如：创建http 服务
restEndpoint, err := endpoint.Registry.New(rest.Type, config, rest.Config{Server: ":9090",})
// 或者使用map方式设置配置
restEndpoint, err := endpoint.Registry.New(rest.Type, config, types.Configuration{"server": ":9090",})

//例如：创建mqtt订阅 服务
mqttEndpoint, err := endpoint.Registry.New(mqtt.Type, config, mqtt.Config{Server: "127.0.0.1:1883",})
// 或者使用map方式设置配置
mqttEndpoint, err := endpoint.Registry.New(mqtt.Type, config, types.Configuration{"server": "127.0.0.1:1883",})

//例如：创建ws服务
wsEndpoint, err := endpoint.Registry.New(websocket.Type, config, websocket.Config{Server: ":9090"})

//例如：创建tcp服务
tcpEndpoint, err := endpoint.Registry.New(net.Type, config, Config{Protocol: "tcp", Server:   ":8888",})

//例如： 创建schedule endpoint服务
scheduleEndpoint, err := endpoint.Registry.New(schedule.Type, config, nil)

3. 把路由注册到endpoint服务中，并启动服务

//http endpoint注册路由
_, err = restEndpoint.AddRouter(router1,"POST")
_, err = restEndpoint.AddRouter(router2,"GET")
_ = restEndpoint.Start()

//mqtt endpoint注册路由
_, err = mqttEndpoint.AddRouter(router1)
_, err = mqttEndpoint.AddRouter(router2)
_ = mqttEndpoint.Start()

4. Endpoint支持响应给调用方

router5 := endpoint.NewRouter().From("/api/v1/msgToComponent2/:msgType").Process(func(router endpoint.Router, exchange *endpoint.Exchange) bool {
    //响应给客户端
    exchange.Out.Headers().Set("Content-Type", "application/json")
    exchange.Out.SetBody([]byte("ok"))
    return true
})
//如果需要把规则链执行结果同步响应给客户端，则增加wait语义
router5 := endpoint.NewRouter().From("/api/v1/msg2Chain4/:chainId").
To("chain:${chainId}").
//必须增加Wait，异步转同步，http才能正常响应，如果不响应同步响应，不要加这一句，会影响吞吐量
Wait().
Process(func(router endpoint.Router, exchange *endpoint.Exchange) bool {
  err := exchange.Out.GetError()
  if err != nil {
    //错误
    exchange.Out.SetStatusCode(400)
    exchange.Out.SetBody([]byte(exchange.Out.GetError().Error()))
    } else {
    //把处理结果响应给客户端，http endpoint 必须增加 Wait()，否则无法正常响应
    outMsg := exchange.Out.GetMsg()
    exchange.Out.Headers().Set("Content-Type", "application/json")
    exchange.Out.SetBody([]byte(outMsg.Data))
  }

  return true
}).End()

5. 添加全局拦截器，用来进行权限校验等逻辑

restEndpoint.AddInterceptors(func(exchange *endpoint.Exchange) bool {
  //权限校验逻辑
  return true
})