@@ -111,56 +111,29 @@ func BodyToResource(body []byte, m Metrics) (*logging.Resource, error) {
111111 return resource , nil
112112}
113113
114- func contains (s []int , e int ) bool {
115- for _ , a := range s {
116- if a == e {
117- return true
118- }
119- }
120- return false
121- }
122-
123114func (pl * Deliverer ) flushBatch (ctx context.Context , resources []logging.Resource , count int , queue Queue ) (int , error ) {
124115 tracer := opentracing .GlobalTracer ()
125116 span , _ := opentracing .StartSpanFromContextWithTracer (ctx , tracer , "deliverer_flush_batch" )
126117 defer span .Finish ()
127118 fmt .Printf ("batch flushing %d messages\n " , count )
128- maxLoop := count
129- l := 0
130119
131- for {
132- l ++
133- resp , err := pl .storer .StoreResources (resources , count )
134- if err == nil { // Happy flow
135- break
136- }
120+ resp , err := pl .storer .StoreResources (resources , count )
121+ if err != nil { // Unpack and send individually
137122 if resp == nil {
138123 fmt .Printf ("unexpected error for StoreResource(): %v\n " , err )
139- continue
140- }
141- nrErrors := len (resp .Failed )
142- keys := make ([]int , 0 , nrErrors )
143- for k := range resp .Failed {
144- keys = append (keys , k )
124+ return count , err
145125 }
146- // Remove offending messages and resend
147- pos := 0
126+ // Unpack and send individual messages
148127 for i := 0 ; i < count ; i ++ {
149- if contains (keys , i ) {
128+ fmt .Printf ("resending %d\n " , i + 1 )
129+ _ , err = pl .storer .StoreResources ([]logging.Resource {resources [i ]}, 1 )
130+ if err != nil {
150131 _ = queue .DeadLetter (resources [i ])
151- continue
132+ fmt . Printf ( "permanent failure sending %d resource: [%v] error: %v \n " , i + 1 , resources [ i ], err )
152133 }
153- resources [pos ] = resources [i ]
154- pos ++
155- }
156- count = pos
157- fmt .Printf ("Found %d errors. Resending %d\n " , nrErrors , count )
158-
159- if l > maxLoop || count <= 0 {
160- fmt .Printf ("Maximum retries reached or nothingt to send. Bailing..\n " )
161- break
162134 }
163135 }
136+
164137 return count , nil
165138}
166139
0 commit comments