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

Question

我正在尝试处理有关 Channels 的使用和正确处理的问题(很可能是设计问题)。
我正在使用 Knative Eventing/Cloud Events 来创建和事件管道。

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

为了做到这一点，我有下面的实现(为了简洁地详细说明问题，代码已被删除)。

我有一个 file1.go它定义了 EventHandler结构、关联方法和几个导出方法(CreatePreview() 和 SaveAndPublish())是应用程序的“正常”行为，并且实际上接收/处理 channel 上的任何值:

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 , 为了使用 eventHandler.Handle c.StartReceiver()上的方法在 main.go我可以定义一个包装器，它可能包含一个事件处理程序列表( HandlerWrapper() 上的 main.go 方法)。

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

我该如何处理这种情况，我想创建一个 Wrapper(一个传递给 c.StartReceive() 的函数)，然后让它由 Handle() 的正确实例处理?

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

Answer 1

据推测，您应该能够通过使用来自该事件的不同来源/方法来区分事件。快速浏览event spec例如，显示您可以根据来源分成 channel 。

我看到这里没有使用的主要内容是上下文对象。看来您可以从该上下文中收集来源。这可以在他们的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实例化那些不同的 channel 和所有这些实现的接口(interface)。

// 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
    }()
}