func()

in plugins/forwarder/grpc/nativemeter/forwarder.go [91:111]

• 19 lines of code
• 6 McCabe index (conditional complexity)

func (f *Forwarder) Forward(batch event.BatchEvents) error {
	streamMap := make(map[string]grpc.ClientStream)
	defer func() {
		for _, stream := range streamMap {
			err := closeStream(stream)
			if err != nil {
				log.Logger.Warnf("%s close stream error: %v", f.Name(), err)
			}
		}
	}()
	for _, e := range batch {
		// Only handle the meter collection data from queue
		// There could have error when using previously meter data(SniffData_Meter)
		if data, ok := e.GetData().(*v1.SniffData_MeterCollection); ok {
			if err := f.handleMeterCollection(data, streamMap); err != nil {
				return err
			}
		}
	}
	return nil
}