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

Question

我正在尝试处理有关 Channel 的使用和正确处理的问题（很可能是设计问题）。 我正在使用 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
}

现在，我知道不必实例化单个全局 eventHandler，为了在 c.StartReceiver() 上使用 eventHandler.Handle 方法 main.go 我可以定义一个包装器，它可能包含一个eventHandler 列表（main.go 上的 HandlerWrapper() 方法）。

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

如果我想创建一个 Wrapper（传递给 c.StartReceive() 的单个函数），然后让它由 Handle() 的正确实例处理，我该如何处理？

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

Answer 1

大概，您应该能够通过使用来自该事件的不同来源/方法来区分事件。例如，快速查看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
    }()
}