0

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

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

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

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

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

4

1 回答 1

1

据推测，您应该能够通过使用来自该事件的不同来源/方法来区分事件。例如，快速查看事件规范表明您可以根据源拆分为通道。

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

// 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
    }()
}
于 2019-11-19T01:40:42.813 回答