in plugins/forwarder/grpc/nativeprofile/forwarder.go [73:95]
func (f *Forwarder) Forward(batch event.BatchEvents) error {
stream, err := f.profileClient.CollectSnapshot(context.Background())
if err != nil {
log.Logger.Errorf("open grpc stream error %v", err)
return err
}
for _, e := range batch {
data, ok := e.GetData().(*v1.SniffData_Profile)
if !ok {
continue
}
err := stream.Send(data.Profile)
if err != nil {
log.Logger.Errorf("%s send log data error: %v", f.Name(), err)
err = closeStream(stream)
if err != nil {
log.Logger.Errorf("%s close stream error: %v", f.Name(), err)
}
return err
}
}
return closeStream(stream)
}