| 144 | } |
| 145 | |
| 146 | func (s *TCP) handleStream(conn *net.TCPConn) error { |
| 147 | r, err := gzip.NewReader(conn) |
| 148 | if err != nil { |
| 149 | return fmt.Errorf("error initializing gzip: %v", err) |
| 150 | } |
| 151 | defer r.Close() |
| 152 | |
| 153 | rbuf := bufio.NewReaderSize(r, tcpChunkBuffer) |
| 154 | |
| 155 | for atomic.LoadInt64(&s.stop) == 0 { |
| 156 | bakerData := s.pool.Get().(*baker.Data) |
| 157 | |
| 158 | // Read a big chunk of data (but keeping tcpMaxLineLength |
| 159 | // bytes available for completing the last line). |
| 160 | n, err := rbuf.Read(bakerData.Bytes[:tcpChunkBuffer-tcpMaxLineLength]) |
| 161 | if err == io.EOF { |
| 162 | bakerData.Bytes = bakerData.Bytes[:n] |
| 163 | s.send(bakerData) |
| 164 | break |
| 165 | } |
| 166 | |
| 167 | if err != nil { |
| 168 | return fmt.Errorf("error reading stream: %v", err) |
| 169 | } |
| 170 | |
| 171 | // We need to send a batch of complete lines to the filter |
| 172 | // (sending truncated lines would generate parsing errors), |
| 173 | // so we want to finish reading the last line we read until its |
| 174 | // terminator. |
| 175 | // NOTE: it might also happen that the chunk we just read |
| 176 | // finished the file; so we check if the chunk ends with a |
| 177 | // terminator, to avoid receiving a io.EOF from ReadBytes; EOFs |
| 178 | // will be handled back when we begin the loop again. |
| 179 | if bakerData.Bytes[n-1] != '\n' { |
| 180 | endl, err := rbuf.ReadBytes('\n') |
| 181 | if err != nil { |
| 182 | return fmt.Errorf("error searching for new line char: %v", err) |
| 183 | } |
| 184 | |
| 185 | // If there is no space in the buffer to complete the |
| 186 | // current line, we need to handle it differently. |
| 187 | if n+len(endl) > tcpChunkBuffer { |
| 188 | // Drop the initial part of the truncated line from the buffer |
| 189 | lastn := n |
| 190 | n = bytes.LastIndexByte(bakerData.Bytes[:n], '\n') + 1 |
| 191 | |
| 192 | // Process the huge line by itself. Allocate a new buffer |
| 193 | // from the pool, copy the initial part, and then concatenate |
| 194 | // up to the endline |
| 195 | bakerData2 := s.pool.Get().(*baker.Data) |
| 196 | bakerData2.Meta = bakerData.Meta |
| 197 | bakerData2.Bytes = append(bakerData2.Bytes[:0], bakerData.Bytes[n:lastn]...) |
| 198 | bakerData2.Bytes = append(bakerData2.Bytes, endl...) |
| 199 | s.send(bakerData2) |
| 200 | } else { |
| 201 | copy(bakerData.Bytes[n:], endl) |
| 202 | n += len(endl) |
| 203 | } |