为多个实例创建事件处理程序封装器

偷偷努力，悄无声息地变强，然后惊艳所有人！哈哈，小伙伴们又来学习啦~今天我将给大家介绍《为多个实例创建事件处理程序封装器》，这篇文章主要会讲到等等知识点，不知道大家对其都有多少了解，下面我们就一起来看一吧！当然，非常希望大家能多多评论，给出合理的建议，我们一起学习，一起进步！

我正在尝试解决有关通道的使用和正确处理的问题（很可能是设计问题）。 我正在使用 knative eventing/cloud events 来创建事件管道。

我希望能够处理不同的通道，以便接收来自不同来源/方法的事件。

为了做到这一点，我有以下实现（代码已被删除，以便简洁地详细说明问题）。

我有一个 file1.go ，它定义了 eventhandler 结构、关联方法和几个导出方法（createpreview() 和 saveandpublish()），它们是应用程序的“正常”行为，并且实际接收/处理频道上出现的任何价值：

type eventhandler struct {
  channel chan string
}

func (ev *eventhandler) handle(event cloudevents.event) {
  if event.data == nil {
    (...)
  }

  var data string
  if err := event.dataas(&data); err != nil {
    (...)
  }

  ev.channel <- data

  defer close(ev.channel)
}

func (ev *eventhandler) create(param *element) (error) {
  (...) //unimportant code
}

func (repo *repository) createpreview(param1 string, param2 string, eventhandler *eventhandler) (*pb.previewresponse, error) {
  (...)
  err := eventhandler.create(&document)
  (...)
  preview := <- eventhandler.channel
  (...)
}

func (repo *repository) saveandpublish(param1 string, param2 bool, eventhandler *eventhandler) (*pb.publishresponse, error) {

  (...)
  err := eventhandler.create(&documenttoupdate)
  (...)

  published := <- eventhandler.channel
  (...)
  return repo.someothermethod(published.id)
}

现在，在我的 main.go 函数上，我可以“常规”启动服务，包括 grpc 监听器、http 监听器和事件处理。这是通过 cmux 完成的。这是代码示例（再次简化代码）：

func HandlerWrapper(event cloudevents.Event) {
    //TODO: HOW DO I HANDLE THIS???
}

// This approach seems to cause issues since it's always the same instance
// var (
//  eventHandler = &rep.EventHandler{Channel: make(chan string)}
// )

func (service *RPCService) Start() (err error) {

  port, err := strconv.Atoi(os.Getenv("LISTEN_PORT"))
  if err != nil {
    log.Fatalf("failed to listen: %v", err)
  }

  lis, err := net.Listen("tcp", fmt.Sprintf(":%d", port))
  if err != nil {
    log.Fatalf("failed to listen: %v", err)
  }

  // Create multiplexer and listener types
  mux := cmux.New(lis)
  grpcLis := mux.Match(cmux.HTTP2())
  httpLis := mux.Match(cmux.HTTP1())

  // *************
  // gRPC
  // *************
  service.server = grpc.NewServer()
  reflection.Register(service.server)
  pb.RegisterStoryServiceServer(service.server, service)

  // *************
  // HTTP
  // *************

  // Declare new CloudEvents Receiver
  c, err := kncloudevents.NewDefaultClient(httpLis)
  if err != nil {
    log.Fatal("Failed to create client, ", err)
  }

  // *************
  // Start Listeners
  // *************

  // start gRPC server
  go func() {
    if err := service.server.Serve(grpcLis); err != nil {
      log.Fatalf("failed to gRPC serve: %s", err)
    }
  }()

  // start HTTP server
  go func() {
    // With this line bellow, I'd have to create a Received per eventHandler. Not cool
    // log.Fatal(c.StartReceiver(context.Background(), eventHandler.Handle))

    // Here we use a wrapper to deal with the event handling and have a single listener
    log.Fatal(c.StartReceiver(context.Background(), HandlerWrapper))
  }()

  if err := mux.Serve(); err != nil {
    log.Fatalf("failed to Mux serve: %s", err)
  }

  return
}

//CreatePreview is used to save a preview for a story 
func (service *RPCService) CreatePreview(ctx context.Context, input *pb.PreviewRequest) (*pb.PreviewResponse, error){

  eventHandler := &rep.EventHandler{Channel: make(chan string)}

  story, err := service.repo.CreatePreview("param1", "param2", eventHandler)
  if err != nil {
    return nil, err
  }

  // Return matching the `CreatePreviewResponse` message we created in our
  // protobuf definition.
  return &pb.PreviewResponse{Story: story}, nil
}

// SaveAndPublish is used to save a story and publish it, returning the story saved. 
func (service *RPCService) SaveAndPublish(ctx context.Context, input *pb.PublishRequest) (*pb.PublishResponse, error){

  eventHandler := &rep.EventHandler{Channel: make(chan string)}

  story, err := service.repo.SaveAndPublish("param1", true, eventHandler)

  if err != nil {
    return nil, err
  }

  // Return matching the `SaveAndPublishResponse` message we created in our
  // protobuf definition.
  return &pb.PublishResponse{Story: story}, nil
}

现在，我知道，为了在 c.startreceiver() 上的 main.go 上使用 eventhandler.handle 方法，我可以定义一个包装器，该包装器可能包含eventhandler 列表（main.go 上的 handlerwrapper() 方法）。

但是，我不知道如何识别 eventhandler 的哪个实例以及如何正确处理和路由这些操作，这就是我的问题：

如果我想创建一个包装器（传递到 c.startreceive() 的单个函数），然后让它由 handle() 的正确实例处理，我该如何处理？

我希望问题很清楚。几天来我一直在尝试解决这个问题，但不知道该怎么做。

解决方案

据推测，您应该能够通过使用来自该事件的不同来源/方法来区分事件。例如，快速浏览一下 event spec 就会发现您可以根据来源分成多个频道。

我发现这里没有使用的主要内容是上下文对象。看来您可以从该上下文中收集来源。这可以在他们的 hello world 示例中看到（查看接收函数）。

举个例子：

// these are all the handlers for the different sources.
type eventhandlers map[string]cloudeventhandler

var _eventhandlerkey = "cloudeventhandlers"

func handlerwrapper(ctx context.context, event cloudevents.event) {
    // get event source from event context.
    src := event.context.source
    // then get the appropriate handler for that source (attached to context).
    handler := ctx.value(_eventhandlers).(values)[src]
    // ex: src = "/foo" 
    handler.sendtochannel(event)
}

func main() {
    eventhandlers := make(map[string]cloudeventhandler)
    // create all the channels we need, add it to the context.
    for _, source := range sourcetypes { // foo bar baz
        handler := newhandler(source)
        eventhandlers[source] = handler
    }

  // start http server
  go func() {
    // add the handlers to the context.
    context := context.withvalue(context.background(), _eventhandlerkey, eventhandlers)
    log.fatal(c.startreceiver(context.background(), handlerwrapper))
  }
}()

如果要支持 3 个不同的源，您可以使用 factory pattern 实例化这些不同的通道以及所有这些通道都实现的接口。

// CloudEventHandler Handles sending cloud events to the proper channel for processing.
type CloudEventHandler interface {
    SendToChannel(cloudEvents.Event)
}

type fooHandler struct {channel chan string}
type barHandler struct {channel chan int}
type bazHandler struct {channel chan bool}

func NewHandler(source string) CloudEventHandler {
    switch source {
    case "/foo":
        return &fooHandler{channel: make(chan string, 2)} // buffered channel
    case "/bar":
        return &barHandler{channel: make(chan int, 2)}
    case "/baz":
        return &bazHandler{channel: make(chan bool, 2)}
    }
}

func (fh *fooHandler) SendToChannel(event CloudEvents.Event) {
    var data string
    if err := event.DataAs(&data); err != nil {
        // (...)
    }

    go func() {
        fh.channel <- data
    }()
}

func (bh *barHandler) SendToChannel(event CloudEvents.Event) {
    var data int
    if err := event.DataAs(&data); err != nil {
        // (...)
    }

    go func() {
        bh.channel <- data
    }()
}

到这里，我们也就讲完了《为多个实例创建事件处理程序封装器》的内容了。个人认为，基础知识的学习和巩固，是为了更好的将其运用到项目中，欢迎关注golang学习网公众号，带你了解更多关于的知识点！