Skip to content

Commit 402a22f

Browse files
Shahrin23Shahrin23
committed
Reduce cognitive complexity
1 parent 51ed8ea commit 402a22f

5 files changed

Lines changed: 390 additions & 406 deletions

File tree

‎core/src/main/java/io/questdb/client/cutlass/http/client/AbstractChunkedResponse.java‎

Lines changed: 188 additions & 121 deletions
Original file line numberDiff line numberDiff line change
@@ -1,24 +1,24 @@
11
/*+*****************************************************************************
2-
* ___ _ ____ ____
3-
* / _ \ _ _ ___ ___| |_| _ \| __ )
4-
* | | | | | | |/ _ \/ __| __| | | | _ \
5-
* | |_| | |_| | __/\__ \ |_| |_| | |_) |
6-
* \__\_\\__,_|\___||___/\__|____/|____/
2+
* ___ _ ____ ____
3+
* / _ \ _ _ ___ ___| |_| _ \| __ )
4+
* | | | | | | |/ _ \/ __| __| | | | _ \
5+
* | |_| | |_| | __/\__ \ |_| |_| | |_) |
6+
* \__\_\\__,_|\___||___/\__|____/|____/
77
*
8-
* Copyright (c) 2014-2019 Appsicle
9-
* Copyright (c) 2019-2026 QuestDB
8+
* Copyright (c) 2014-2019 Appsicle
9+
* Copyright (c) 2019-2026 QuestDB
1010
*
11-
* Licensed under the Apache License, Version 2.0 (the "License");
12-
* you may not use this file except in compliance with the License.
13-
* You may obtain a copy of the License at
11+
* Licensed under the Apache License, Version 2.0 (the "License");
12+
* you may not use this file except in compliance with the License.
13+
* You may obtain a copy of the License at
1414
*
15-
* http://www.apache.org/licenses/LICENSE-2.0
15+
* http://www.apache.org/licenses/LICENSE-2.0
1616
*
17-
* Unless required by applicable law or agreed to in writing, software
18-
* distributed under the License is distributed on an "AS IS" BASIS,
19-
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
20-
* See the License for the specific language governing permissions and
21-
* limitations under the License.
17+
* Unless required by applicable law or agreed to in writing, software
18+
* distributed under the License is distributed on an "AS IS" BASIS,
19+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
20+
* See the License for the specific language governing permissions and
21+
* limitations under the License.
2222
*
2323
******************************************************************************/
2424

@@ -90,125 +90,175 @@ public long lo() {
9090
return dataAddr;
9191
}
9292

93+
/**
94+
* Reads the next fragment of chunked response data, blocking on I/O as needed.
95+
* <p>
96+
* This method drives a small state machine with three states: reading the
97+
* chunk-size header, consuming chunk data, and consuming the chunk terminator.
98+
* Each state's logic is delegated to a dedicated helper method to keep this
99+
* driver method simple and easy to follow.
100+
*/
93101
public Fragment recv(int timeout) {
94102
while (true) {
95103
if (receive || dataLo == dataHi) {
96104
compactBuffer();
97105
dataHi += recvOrDie(dataHi, bufHi, timeout);
98106
}
99-
long p; // moving data pointer for scanning buffer
100-
switch (state) {
101-
case STATE_CHUNK_SIZE:
102-
p = dataLo;
103-
// chunk size is hex encoded integer terminated with CRLF
104-
long res = -1;
105-
106-
// this loop is looking at the CRLF after chunk size
107-
while (p < dataHi) {
108-
if (getByte(p) == '\r') {
109-
p++;
110-
if (p < dataHi) {
111-
if (getByte(p) == '\n') {
112-
res = p - CRLF_LEN;
113-
break;
114-
} else {
115-
throw new HttpClientException("malformed chunk size");
116-
}
117-
} else {
118-
// CRLF at chunk size is incomplete, we have to
119-
// wait until we receive the full thing
120-
break;
121-
}
122-
}
123-
p++;
124-
}
125-
126-
if (res != -1) {
127-
// at this stage we consumed the chunk size end (CRLF)
128-
chunkSize.of(dataLo, res + 1);
129-
try {
130-
size = Numbers.parseHexLong(chunkSize.asAsciiCharSequence());
131-
consumed = 0;
132-
// consume data buffer ignoring chunk size value and its furniture
133-
state = STATE_CHUNK_DATA;
134-
dataLo = res + CRLF_LEN + 1;
135-
} catch (NumericException e) {
136-
throw new HttpClientException("malformed chunk size");
137-
}
138-
139-
// fall thru the switch to process remaining data buffer
140-
} else {
141-
// we have not received complete chunk size value yet
142-
receive = true;
143-
break;
144-
}
145-
146-
case STATE_CHUNK_DATA:
147-
// there is data in the buffer
148-
if (size > 0 && dataLo < dataHi) {
149-
long chunkBytesRemaining = size - consumed;
150-
long bufBytesRemaining = dataHi - dataLo;
151-
152-
// chunk data starts with dataLo address
153-
dataAddr = dataLo;
154-
155-
if (chunkBytesRemaining <= bufBytesRemaining) {
156-
// chunk data fits in the buffer
157-
available = chunkBytesRemaining;
158-
consumed += chunkBytesRemaining;
159-
// skip chunk data to begin processing chunk end
160-
dataLo += chunkBytesRemaining;
161-
state = STATE_CHUNK_DATA_END;
162-
receive = false;
163-
} else {
164-
available = bufBytesRemaining;
165-
consumed += bufBytesRemaining;
166-
// we consumed the entire buffer for chunk data
167-
// we must recv more data
168-
dataLo = dataHi;
169-
receive = true;
170-
}
171-
return this;
172-
}
173-
174-
if (size != 0) {
175-
// no chunk data in the buffer
176-
break;
177-
}
178-
// fall thru to read chunk end
179-
180-
case STATE_CHUNK_DATA_END:
181-
// we are here to consume CRLF
182-
// we have to have two bytes here
183-
if (dataLo < dataHi && (dataHi - dataLo) >= CRLF_LEN) {
184-
if (getByte(dataLo) == '\r' && getByte(dataLo + 1) == '\n') {
185-
state = STATE_CHUNK_SIZE;
186-
dataLo += CRLF_LEN;
187-
receive = false;
188-
// we had to consume the tail CRLF after the last chunk
189-
// not to leave garbage in the recv buffer
190-
if (size == 0) {
191-
return null;
192-
}
193-
break;
194-
} else {
195-
throw new HttpClientException("malformed chunk");
196-
}
197-
} else {
198-
receive = true;
199-
}
200-
break;
201-
default:
202-
throw new HttpClientException("internal error [state=" + state + ']');
107+
108+
StateOutcome outcome = processState();
109+
if (outcome == StateOutcome.RETURN_FRAGMENT) {
110+
return this;
111+
}
112+
if (outcome == StateOutcome.RETURN_NULL) {
113+
return null;
203114
}
204115
}
205116
}
206117

118+
private StateOutcome processState() {
119+
switch (state) {
120+
case STATE_CHUNK_SIZE:
121+
if (!tryParseChunkSize()) {
122+
receive = true;
123+
return StateOutcome.CONTINUE;
124+
}
125+
// fall through: chunk size parsed, state is now STATE_CHUNK_DATA
126+
127+
case STATE_CHUNK_DATA:
128+
ChunkDataOutcome dataOutcome = tryConsumeChunkData();
129+
if (dataOutcome == ChunkDataOutcome.FRAGMENT_READY) {
130+
return StateOutcome.RETURN_FRAGMENT;
131+
}
132+
if (dataOutcome == ChunkDataOutcome.NEED_MORE_DATA) {
133+
return StateOutcome.CONTINUE;
134+
}
135+
// dataOutcome == CHUNK_EXHAUSTED: fall through to consume the terminator
136+
137+
case STATE_CHUNK_DATA_END:
138+
if (tryConsumeChunkEnd() == ChunkEndOutcome.END_OF_STREAM) {
139+
return StateOutcome.RETURN_NULL;
140+
}
141+
return StateOutcome.CONTINUE;
142+
143+
default:
144+
throw new HttpClientException("internal error [state=" + state + ']');
145+
}
146+
}
147+
207148
@Override
208149
public Fragment recv() {
209150
return recv(defaultTimeout);
210151
}
211152

153+
/**
154+
* Attempts to locate and parse the hex-encoded chunk-size header terminated by CRLF.
155+
*
156+
* @return {@code true} if the chunk size was fully parsed and {@code state} advanced
157+
* to {@code STATE_CHUNK_DATA}; {@code false} if more data must be received
158+
* before the header can be completed
159+
*/
160+
private boolean tryParseChunkSize() {
161+
long terminatorPos = findChunkSizeTerminator();
162+
if (terminatorPos == -1) {
163+
return false;
164+
}
165+
166+
chunkSize.of(dataLo, terminatorPos + 1);
167+
try {
168+
size = Numbers.parseHexLong(chunkSize.asAsciiCharSequence());
169+
} catch (NumericException e) {
170+
throw new HttpClientException("malformed chunk size");
171+
}
172+
consumed = 0;
173+
state = STATE_CHUNK_DATA;
174+
dataLo = terminatorPos + CRLF_LEN + 1;
175+
return true;
176+
}
177+
178+
/**
179+
* Scans the buffer for the CRLF that terminates the chunk-size header.
180+
*
181+
* @return the index of the CRLF start position, or -1 if the buffer does not
182+
* yet contain a complete chunk-size header
183+
*/
184+
private long findChunkSizeTerminator() {
185+
long p = dataLo;
186+
while (p < dataHi) {
187+
if (getByte(p) == '\r') {
188+
p++;
189+
if (p >= dataHi) {
190+
// incomplete CRLF, must wait for more data
191+
return -1;
192+
}
193+
if (getByte(p) != '\n') {
194+
throw new HttpClientException("malformed chunk size");
195+
}
196+
return p - CRLF_LEN;
197+
}
198+
p++;
199+
}
200+
return -1;
201+
}
202+
203+
/**
204+
* Attempts to consume as much chunk data as is currently available in the buffer.
205+
*
206+
* @return {@link ChunkDataOutcome#FRAGMENT_READY} if a data fragment is ready to
207+
* be returned to the caller; {@link ChunkDataOutcome#NEED_MORE_DATA} if the
208+
* buffer must be refilled; {@link ChunkDataOutcome#CHUNK_EXHAUSTED} if the
209+
* current chunk has zero remaining bytes and the terminator should be read
210+
*/
211+
private ChunkDataOutcome tryConsumeChunkData() {
212+
if (size > 0 && dataLo < dataHi) {
213+
long chunkBytesRemaining = size - consumed;
214+
long bufBytesRemaining = dataHi - dataLo;
215+
dataAddr = dataLo;
216+
217+
if (chunkBytesRemaining <= bufBytesRemaining) {
218+
available = chunkBytesRemaining;
219+
consumed += chunkBytesRemaining;
220+
dataLo += chunkBytesRemaining;
221+
state = STATE_CHUNK_DATA_END;
222+
receive = false;
223+
} else {
224+
available = bufBytesRemaining;
225+
consumed += bufBytesRemaining;
226+
dataLo = dataHi;
227+
receive = true;
228+
}
229+
return ChunkDataOutcome.FRAGMENT_READY;
230+
}
231+
232+
if (size != 0) {
233+
return ChunkDataOutcome.NEED_MORE_DATA;
234+
}
235+
236+
return ChunkDataOutcome.CHUNK_EXHAUSTED;
237+
}
238+
239+
/**
240+
* Attempts to consume the CRLF that terminates a chunk's data.
241+
*
242+
* @return {@link ChunkEndOutcome#END_OF_STREAM} if the zero-length terminating
243+
* chunk was consumed; {@link ChunkEndOutcome#NEED_MORE_DATA} otherwise,
244+
* meaning either more data is required or the next chunk is ready to process
245+
*/
246+
private ChunkEndOutcome tryConsumeChunkEnd() {
247+
if (dataLo >= dataHi || (dataHi - dataLo) < CRLF_LEN) {
248+
receive = true;
249+
return ChunkEndOutcome.NEED_MORE_DATA;
250+
}
251+
252+
if (getByte(dataLo) != '\r' || getByte(dataLo + 1) != '\n') {
253+
throw new HttpClientException("malformed chunk");
254+
}
255+
256+
state = STATE_CHUNK_SIZE;
257+
dataLo += CRLF_LEN;
258+
receive = false;
259+
return size == 0 ? ChunkEndOutcome.END_OF_STREAM : ChunkEndOutcome.NEED_MORE_DATA;
260+
}
261+
212262
private void compactBuffer() {
213263
// move unprocessed data to the front of the buffer
214264
// to maximise
@@ -239,4 +289,21 @@ private byte getByte(long addr) {
239289
* @return the number of bytes received
240290
*/
241291
protected abstract int recvOrDie(long bufLo, long bufHi, int timeout);
242-
}
292+
293+
private enum ChunkDataOutcome {
294+
FRAGMENT_READY,
295+
NEED_MORE_DATA,
296+
CHUNK_EXHAUSTED
297+
}
298+
299+
private enum ChunkEndOutcome {
300+
END_OF_STREAM,
301+
NEED_MORE_DATA
302+
}
303+
304+
private enum StateOutcome {
305+
CONTINUE,
306+
RETURN_FRAGMENT,
307+
RETURN_NULL
308+
}
309+
}

0 commit comments

Comments
 (0)