func()

in plugins/forwarder/grpc/nativeebpfaccesslog/forwarder.go [74:105]

• 28 lines of code
• 7 McCabe index (conditional complexity)

func (f *Forwarder) Forward(batch event.BatchEvents) (err error) {
	var stream v3.EBPFAccessLogService_CollectClient
	for _, e := range batch {
		data, ok := e.GetData().(*v1.SniffData_EBPFAccessLogList)
		if !ok {
			continue
		}

		stream, err = f.accessClient.Collect(context.Background())
		if err != nil {
			log.Logger.Errorf("open grpc stream error: %v", err)
			return err
		}

		streamClosed := false
		for _, message := range data.EBPFAccessLogList.Messages {
			err := stream.SendMsg(server_grpc.NewOriginalData(message))
			if err != nil {
				log.Logger.Errorf("%s send log data error: %v", f.Name(), err)
				f.closeStream(stream)
				streamClosed = true
				break
			}
		}

		if !streamClosed {
			f.closeStream(stream)
		}
	}

	return nil
}