@@ -91,40 +91,15 @@ func (c *Component) Stop() error {
91
91
92
92
// Push handles a push request
93
93
func (c * Component ) Push (ctx context.Context , in * pb.PushRequest ) (* pb.PushResponse , error ) {
94
- reqID := in .GetId ()
95
- if reqID == "" {
96
- return nil , fmt .Errorf ("id is empty" )
94
+ if err := validatePushRequest (in ); err != nil {
95
+ return nil , err
97
96
}
98
97
99
98
reqStream := in .GetStream ()
100
99
if reqStream == "" {
101
100
reqStream = DefaultRequestStream
102
101
}
103
102
104
- producerAppName := in .GetProducerAppName ()
105
- if producerAppName == "" {
106
- return nil , fmt .Errorf ("producer app name is empty" )
107
- }
108
-
109
- producerAppVersion := in .GetProducerAppVersion ()
110
- if producerAppVersion == "" {
111
- return nil , fmt .Errorf ("producer app version is empty" )
112
- }
113
-
114
- sdkName := in .GetSdkName ()
115
- if sdkName == "" {
116
- return nil , fmt .Errorf ("sdk name is empty" )
117
- }
118
-
119
- sdkVersion := in .GetSdkVersion ()
120
- if sdkVersion == "" {
121
- return nil , fmt .Errorf ("sdk version is empty" )
122
- }
123
-
124
- if len (in .GetData ()) < 1 {
125
- return nil , fmt .Errorf ("data is empty" )
126
- }
127
-
128
103
errs := make ([]string , 0 )
129
104
130
105
for i , v := range in .GetData () {
@@ -139,12 +114,12 @@ func (c *Component) Push(ctx context.Context, in *pb.PushRequest) (*pb.PushRespo
139
114
continue
140
115
}
141
116
msg := map [string ]interface {}{
142
- "id" : reqID ,
117
+ "id" : in . GetId () ,
143
118
"stream" : reqStream ,
144
- "producer_app_name" : producerAppName ,
145
- "producer_app_version" : producerAppVersion ,
146
- "sdk_name" : sdkName ,
147
- "sdk_version" : sdkVersion ,
119
+ "producer_app_name" : in . GetProducerAppName () ,
120
+ "producer_app_version" : in . GetProducerAppVersion () ,
121
+ "sdk_name" : in . GetSdkName () ,
122
+ "sdk_version" : in . GetSdkVersion () ,
148
123
"data" : encodedEntity ,
149
124
"ts" : v .GetTimestamp ().String (),
150
125
"ingestion_ts" : time .Now ().UnixNano (),
@@ -159,3 +134,31 @@ func (c *Component) Push(ctx context.Context, in *pb.PushRequest) (*pb.PushRespo
159
134
160
135
return & pb.PushResponse {Errors : errs }, nil
161
136
}
137
+
138
+ func validatePushRequest (in * pb.PushRequest ) error {
139
+ if in .GetId () == "" {
140
+ return fmt .Errorf ("id is empty" )
141
+ }
142
+
143
+ if in .GetProducerAppName () == "" {
144
+ return fmt .Errorf ("producer app name is empty" )
145
+ }
146
+
147
+ if in .GetProducerAppVersion () == "" {
148
+ return fmt .Errorf ("producer app version is empty" )
149
+ }
150
+
151
+ if in .GetSdkName () == "" {
152
+ return fmt .Errorf ("sdk name is empty" )
153
+ }
154
+
155
+ if in .GetSdkVersion () == "" {
156
+ return fmt .Errorf ("sdk version is empty" )
157
+ }
158
+
159
+ if len (in .GetData ()) < 1 {
160
+ return fmt .Errorf ("data is empty" )
161
+ }
162
+
163
+ return nil
164
+ }
0 commit comments