@@ -70,7 +70,7 @@ func (r *Reader) ReadToEnd(ctx context.Context) {
7070 // SectionReader can only read a fixed window (from previous offset to EOF).
7171 info , err := r .file .Stat ()
7272 if err != nil {
73- r .set .Logger .Error ("Failed to stat" , zap .Error (err ))
73+ r .set .Logger .Error ("failed to stat" , zap .Error (err ))
7474 return
7575 }
7676 currentEOF := info .Size ()
@@ -80,7 +80,7 @@ func (r *Reader) ReadToEnd(ctx context.Context) {
8080 gzipReader , err := gzip .NewReader (io .NewSectionReader (r .file , r .Offset , currentEOF ))
8181 if err != nil {
8282 if ! errors .Is (err , io .EOF ) {
83- r .set .Logger .Error ("Failed to create gzip reader" , zap .Error (err ))
83+ r .set .Logger .Error ("failed to create gzip reader" , zap .Error (err ))
8484 }
8585 return
8686 } else {
@@ -96,7 +96,7 @@ func (r *Reader) ReadToEnd(ctx context.Context) {
9696 }
9797
9898 if _ , err := r .file .Seek (r .Offset , 0 ); err != nil {
99- r .set .Logger .Error ("Failed to seek" , zap .Error (err ))
99+ r .set .Logger .Error ("failed to seek" , zap .Error (err ))
100100 return
101101 }
102102
@@ -106,9 +106,85 @@ func (r *Reader) ReadToEnd(ctx context.Context) {
106106 }
107107 }()
108108
109+ if r .headerReader != nil {
110+ if r .readHeader (ctx ) {
111+ return
112+ }
113+ }
114+
115+ r .readContents (ctx )
116+ }
117+
118+ func (r * Reader ) readHeader (ctx context.Context ) (doneReadingFile bool ) {
119+ s := scanner .New (r , r .maxLogSize , r .initialBufferSize , r .Offset , r .splitFunc )
120+
121+ // Read the tokens from the file until no more header tokens are found or the end of file is reached.
122+ for {
123+ select {
124+ case <- ctx .Done ():
125+ return true
126+ default :
127+ }
128+
129+ ok := s .Scan ()
130+ if ! ok {
131+ if err := s .Error (); err != nil {
132+ r .set .Logger .Error ("failed during header scan" , zap .Error (err ))
133+ } else {
134+ r .set .Logger .Debug ("end of file reached" , zap .Bool ("delete_at_eof" , r .deleteAtEOF ))
135+ if r .deleteAtEOF {
136+ r .delete ()
137+ }
138+ }
139+ // Either end of file was reached, or file cannot be scanned.
140+ return true
141+ }
142+
143+ token , err := r .decoder .Decode (s .Bytes ())
144+ if err != nil {
145+ r .set .Logger .Error ("failed to decode header token" , zap .Error (err ))
146+ r .Offset = s .Pos () // move past the bad token or we may be stuck
147+ continue
148+ }
149+
150+ err = r .headerReader .Process (ctx , token , r .FileAttributes )
151+ if err != nil {
152+ if errors .Is (err , header .ErrEndOfHeader ) {
153+ // End of header reached.
154+ break
155+ }
156+ r .set .Logger .Error ("failed to process header token" , zap .Error (err ))
157+ }
158+
159+ r .Offset = s .Pos ()
160+ }
161+
162+ // Clean up the header machinery
163+ if err := r .headerReader .Stop (); err != nil {
164+ r .set .Logger .Error ("failed to stop header pipeline during finalization" , zap .Error (err ))
165+ }
166+ r .headerReader = nil
167+ r .HeaderFinalized = true
168+ r .initialBufferSize = scanner .DefaultBufferSize
169+
170+ // Switch to the normal split and process functions.
171+ r .splitFunc = r .lineSplitFunc
172+ r .processFunc = r .emitFunc
173+
174+ // Reset position in file to r.Offest after the header scanner might have moved it past a content token.
175+ if _ , err := r .file .Seek (r .Offset , 0 ); err != nil {
176+ r .set .Logger .Error ("failed to seek post-header" , zap .Error (err ))
177+ return true
178+ }
179+
180+ return false
181+ }
182+
183+ func (r * Reader ) readContents (ctx context.Context ) {
184+ // Create the scanner to read the contents of the file.
109185 s := scanner .New (r , r .maxLogSize , r .initialBufferSize , r .Offset , r .splitFunc )
110186
111- // Iterate over the tokenized file, emitting entries as we go
187+ // Iterate over the contents of the file.
112188 for {
113189 select {
114190 case <- ctx .Done ():
@@ -119,7 +195,7 @@ func (r *Reader) ReadToEnd(ctx context.Context) {
119195 ok := s .Scan ()
120196 if ! ok {
121197 if err := s .Error (); err != nil {
122- r .set .Logger .Error ("Failed during scan" , zap .Error (err ))
198+ r .set .Logger .Error ("failed during scan" , zap .Error (err ))
123199 } else if r .deleteAtEOF {
124200 r .delete ()
125201 }
@@ -128,7 +204,7 @@ func (r *Reader) ReadToEnd(ctx context.Context) {
128204
129205 token , err := r .decoder .Decode (s .Bytes ())
130206 if err != nil {
131- r .set .Logger .Error ("Failed to decode token" , zap .Error (err ))
207+ r .set .Logger .Error ("failed to decode token" , zap .Error (err ))
132208 r .Offset = s .Pos () // move past the bad token or we may be stuck
133209 continue
134210 }
@@ -139,36 +215,11 @@ func (r *Reader) ReadToEnd(ctx context.Context) {
139215 }
140216
141217 err = r .processFunc (ctx , token , r .FileAttributes )
142- if err == nil {
143- r .Offset = s .Pos () // successful emit, update offset
144- continue
145- }
146-
147- if ! errors .Is (err , header .ErrEndOfHeader ) {
148- r .set .Logger .Error ("Failed to process token" , zap .Error (err ))
149- r .Offset = s .Pos () // move past the bad token or we may be stuck
150- continue
218+ if err != nil {
219+ r .set .Logger .Error ("failed to process token" , zap .Error (err ))
151220 }
152221
153- // Clean up the header machinery
154- if err = r .headerReader .Stop (); err != nil {
155- r .set .Logger .Error ("Failed to stop header pipeline during finalization" , zap .Error (err ))
156- }
157- r .headerReader = nil
158- r .HeaderFinalized = true
159-
160- // Switch to the normal split and process functions.
161- r .splitFunc = r .lineSplitFunc
162- r .processFunc = r .emitFunc
163-
164- // Recreate the scanner with the normal split func.
165- // Do not use the updated offset from the old scanner, as the most recent token
166- // could be split differently with the new splitter.
167- if _ , err = r .file .Seek (r .Offset , 0 ); err != nil {
168- r .set .Logger .Error ("Failed to seek post-header" , zap .Error (err ))
169- return
170- }
171- s = scanner .New (r , r .maxLogSize , scanner .DefaultBufferSize , r .Offset , r .splitFunc )
222+ r .Offset = s .Pos ()
172223 }
173224}
174225
0 commit comments