Skip to content

Commit a7c1b64

Browse files
greenlaw110greenlaw110
committed
Error with sending large response: UT000043: Data is already being sent. You must wait for the completion callback to be be invoked before calling send() again actframework#560 
1 parent 57c7177 commit a7c1b64

7 files changed

Lines changed: 173 additions & 34 deletions

File tree

‎CHANGELOG.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
# ActFramework Change Log
22

33
**1.8.2**
4+
* Error with sending large response: UT000043: Data is already being sent. You must wait for the completion callback to be be invoked before calling send() again #560
45
* ResourceLoader: support loading json file into Map or other POJO #559
56
* Support `@DefaultValue` with `@Configuration` #558
67
* `StackOverflowError` encountered when `SimpleBean` field name does not follow Java convention #546

‎pom.xml‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -50,7 +50,7 @@
5050
<cdi-api.version>1.2</cdi-api.version>
5151
<commons-fileupload.version>1.3.3</commons-fileupload.version>
5252
<ecj.version>4.6.1</ecj.version>
53-
<fastjson.version>1.2.46</fastjson.version>
53+
<fastjson.version>1.2.47</fastjson.version>
5454
<bval.version>1.1.2</bval.version>
5555
<image4j.version>0.7</image4j.version>
5656
<jansi.version>1.16</jansi.version>
@@ -66,7 +66,7 @@
6666
<osgl-cache.version>1.3.0</osgl-cache.version>
6767
<osgl-genie.version>1.6.1-SNAPSHOT</osgl-genie.version>
6868
<osgl-http.version>1.4.0</osgl-http.version>
69-
<osgl-mvc.version>1.5.1</osgl-mvc.version>
69+
<osgl-mvc.version>1.5.2-SNAPSHOT</osgl-mvc.version>
7070
<osgl-storage.version>1.5.0</osgl-storage.version>
7171
<osgl-tool-ext.version>1.1.0</osgl-tool-ext.version>
7272
<osgl-ut.version>2.0.0-BETA-1</osgl-ut.version>

‎src/main/java/act/ActResponse.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,8 @@
2424
import act.conf.AppConfig;
2525
import org.osgl.$;
2626
import org.osgl.http.H;
27+
import org.osgl.logging.LogManager;
28+
import org.osgl.logging.Logger;
2729
import org.osgl.mvc.MvcConfig;
2830
import org.osgl.mvc.result.Ok;
2931
import org.osgl.mvc.result.Redirect;

‎src/main/java/act/util/JsonUtilConfig.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,8 @@
2020
* #L%
2121
*/
2222

23+
import static com.alibaba.fastjson.JSON.DEFAULT_GENERATE_FEATURE;
24+
2325
import act.app.App;
2426
import act.cli.util.MappedFastJsonNameFilter;
2527
import act.data.DataPropertyRepository;
@@ -47,8 +49,6 @@
4749
import java.util.List;
4850
import java.util.Set;
4951

50-
import static com.alibaba.fastjson.JSON.DEFAULT_GENERATE_FEATURE;
51-
5252
public class JsonUtilConfig {
5353

5454
public static class JsonWriter extends $.Visitor<org.osgl.util.Output> {

‎src/main/java/act/view/FilteredRenderJSON.java‎

Lines changed: 39 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -25,18 +25,18 @@
2525
import act.util.JsonUtilConfig;
2626
import act.util.PropertySpec;
2727
import org.osgl.$;
28-
import org.osgl.Osgl;
2928
import org.osgl.http.H;
30-
import org.osgl.mvc.MvcConfig;
29+
import org.osgl.http.Http;
3130
import org.osgl.mvc.result.RenderContent;
31+
import org.osgl.mvc.result.RenderJSON;
3232
import org.osgl.util.Output;
3333

3434
/**
3535
* An enhanced version of {@link org.osgl.mvc.result.RenderJSON} that
3636
* allows {@link act.util.PropertySpec} to be applied to control the
3737
* output fields
3838
*/
39-
public class FilteredRenderJSON extends RenderContent {
39+
public class FilteredRenderJSON extends RenderJSON {
4040

4141
public static final FilteredRenderJSON _INSTANCE = new FilteredRenderJSON() {
4242
@Override
@@ -45,31 +45,59 @@ public String content() {
4545
}
4646

4747
@Override
48-
public Osgl.Visitor<Output> contentWriter() {
48+
public $.Visitor<Output> contentWriter() {
4949
return payload().contentWriter;
5050
}
5151

52+
@Override
53+
public Http.Status status() {
54+
Http.Status status = payload().status;
55+
return null == status ? super.status() : status;
56+
}
57+
5258
@Override
5359
public long timestamp() {
5460
return payload().timestamp;
5561
}
62+
63+
@Override
64+
public boolean isOutputEncoding() {
65+
return payload().outputEncoding();
66+
}
67+
68+
@Override
69+
public RenderContent setOutputEncoding(boolean outputEncoding) {
70+
payload().outputEncoding(outputEncoding);
71+
return this;
72+
}
73+
74+
@Override
75+
public H.Format format() {
76+
return H.Format.JSON;
77+
}
5678
};
5779

5880
private FilteredRenderJSON() {
5981
super(H.Format.JSON);
6082
}
6183

6284
public FilteredRenderJSON(final Object v, final PropertySpec.MetaInfo spec, final ActContext context) {
63-
super(new JsonUtilConfig.JsonWriter(v, spec, false, context), H.Format.JSON);
85+
super(new JsonUtilConfig.JsonWriter(v, spec, false, context));
6486
}
6587

66-
public static FilteredRenderJSON get(final Object v, final PropertySpec.MetaInfo spec, final ActionContext context) {
67-
touchPayload().contentWriter(new JsonUtilConfig.JsonWriter(v, spec, false, context));
68-
return _INSTANCE;
88+
public FilteredRenderJSON(H.Status status, final Object v, final PropertySpec.MetaInfo spec, final ActContext context) {
89+
super(status, new JsonUtilConfig.JsonWriter(v, spec, false, context));
6990
}
7091

71-
public FilteredRenderJSON(H.Status status, final Object v, final PropertySpec.MetaInfo spec, final ActContext context) {
72-
super(status, new JsonUtilConfig.JsonWriter(v, spec, false, context), H.Format.JSON);
92+
public static FilteredRenderJSON get(final Object v, final PropertySpec.MetaInfo spec, final ActionContext context) {
93+
if (v instanceof String) {
94+
touchPayload().message((String) v);
95+
} else if (v instanceof $.Visitor) {
96+
touchPayload().contentWriter(($.Visitor) v);
97+
} else {
98+
touchPayload().contentWriter(new JsonUtilConfig.JsonWriter(v, spec, false, context));
99+
}
100+
return _INSTANCE;
73101
}
74102

75103
public static FilteredRenderJSON get(H.Status status, final Object v, final PropertySpec.MetaInfo spec, final ActionContext context) {
@@ -78,7 +106,7 @@ public static FilteredRenderJSON get(H.Status status, final Object v, final Prop
78106
} else if (v instanceof $.Visitor) {
79107
touchPayload().status(status).contentWriter(($.Visitor) v);
80108
} else {
81-
touchPayload().status(status).contentWriter(MvcConfig.jsonSerializer(v));
109+
touchPayload().status(status).contentWriter(new JsonUtilConfig.JsonWriter(v, spec, false, context));
82110
}
83111
return _INSTANCE;
84112
}

‎src/main/java/act/xio/undertow/UndertowResponse.java‎

Lines changed: 125 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -9,9 +9,9 @@
99
* Licensed under the Apache License, Version 2.0 (the "License");
1010
* you may not use this file except in compliance with the License.
1111
* You may obtain a copy of the License at
12-
*
12+
*
1313
* http://www.apache.org/licenses/LICENSE-2.0
14-
*
14+
*
1515
* Unless required by applicable law or agreed to in writing, software
1616
* distributed under the License is distributed on an "AS IS" BASIS,
1717
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@@ -33,6 +33,8 @@
3333
import org.osgl.$;
3434
import org.osgl.exception.UnexpectedIOException;
3535
import org.osgl.http.H;
36+
import org.osgl.logging.LogManager;
37+
import org.osgl.logging.Logger;
3638
import org.osgl.storage.ISObject;
3739
import org.osgl.util.E;
3840
import org.osgl.util.IO;
@@ -43,17 +45,122 @@
4345
import java.io.OutputStream;
4446
import java.nio.ByteBuffer;
4547
import java.nio.channels.FileChannel;
48+
import java.nio.charset.StandardCharsets;
4649
import java.util.Locale;
50+
import java.util.concurrent.locks.Condition;
51+
import java.util.concurrent.locks.ReentrantLock;
4752

4853
public class UndertowResponse extends ActResponse<UndertowResponse> {
4954

55+
protected static Logger LOGGER = LogManager.get(UndertowResponse.class);
56+
57+
private static class Buffer {
58+
boolean isSending;
59+
ByteBuffer buffer;
60+
IoCallback callback;
61+
ReentrantLock lock;
62+
Condition finalPartToGo;
63+
64+
Buffer(IoCallback callback, ReentrantLock lock) {
65+
this.callback = $.notNull(callback);
66+
this.isSending = true;
67+
this.lock = lock;
68+
}
69+
70+
void sendOrBuf(String content, Sender sender) {
71+
sendOrBuf(ByteBuffer.wrap(content.getBytes(StandardCharsets.UTF_8)), sender);
72+
}
73+
74+
void sendOrBuf(ByteBuffer content, Sender sender) {
75+
lock.lock();
76+
try {
77+
if (null == buffer) {
78+
buffer = content;
79+
} else {
80+
ByteBuffer merged = ByteBuffer.allocate(buffer.limit() + content.limit());
81+
merged.put(buffer).put(content).flip();
82+
buffer = merged;
83+
}
84+
if (!isSending) {
85+
isSending = true;
86+
ByteBuffer buffer = this.buffer;
87+
this.buffer = null;
88+
sender.send(buffer, callback);
89+
}
90+
} finally {
91+
lock.unlock();
92+
}
93+
}
94+
95+
void sendThroughFinalPart(Sender sender) {
96+
if (null == this.buffer) {
97+
return;
98+
}
99+
lock.lock();
100+
try {
101+
if (!isSending) {
102+
isSending = true;
103+
sender.send(this.buffer);
104+
} else {
105+
finalPartToGo = lock.newCondition();
106+
while (isSending) {
107+
try {
108+
finalPartToGo.await();
109+
} catch (InterruptedException e) {
110+
Thread.currentThread().interrupt();
111+
throw E.unexpected(e);
112+
}
113+
}
114+
ByteBuffer buffer = this.buffer;
115+
this.buffer = null;
116+
sender.send(buffer, callback);
117+
}
118+
} finally {
119+
lock.unlock();
120+
}
121+
}
122+
123+
void partSent() {
124+
isSending = false;
125+
if (null != finalPartToGo) {
126+
finalPartToGo.signal();
127+
}
128+
}
129+
130+
private void clear() {
131+
isSending = false;
132+
buffer = null;
133+
}
134+
}
135+
50136
private static final HttpString _SERVER = new HttpString(H.Header.Names.SERVER);
51137
private static final HttpStringCache HEADER_NAMES = HttpStringCache.HEADER;
52138

53139
private HttpServerExchange hse;
54140

55141
private boolean endAsync;
56142
private Sender sender;
143+
private ReentrantLock lock;
144+
private IoCallback ioCallback = new DefaultIoCallback() {
145+
@Override
146+
public void onComplete(HttpServerExchange exchange, Sender sender) {
147+
if (null != lock) {
148+
lock.lock();
149+
buffer.partSent();
150+
lock.unlock();
151+
}
152+
}
153+
154+
@Override
155+
public void onException(HttpServerExchange exchange, Sender sender, IOException exception) {
156+
if (null != buffer) {
157+
buffer.clear();
158+
buffer = null;
159+
}
160+
super.onException(exchange, sender, exception);
161+
}
162+
};
163+
private Buffer buffer;
57164

58165
public UndertowResponse(HttpServerExchange exchange, AppConfig config) {
59166
super(config);
@@ -106,17 +213,21 @@ public UndertowResponse writeContent(String s) {
106213
}
107214

108215
public void writeContentPart(String s) {
109-
try {
110-
sender().send(s, WRITE_PART_CALLBACK);
111-
} catch (RuntimeException e) {
112-
endAsync = false;
113-
throw e;
114-
}
216+
ByteBuffer buffer = ByteBuffer.wrap(s.getBytes(StandardCharsets.UTF_8));
217+
writeContentPart(buffer);
115218
}
116219

117220
public void writeContentPart(ByteBuffer buffer) {
118221
try {
119-
sender().send(buffer, WRITE_PART_CALLBACK);
222+
if (null != buffer) {
223+
sender().send(buffer, ioCallback);
224+
} else {
225+
this.buffer.sendOrBuf(buffer, sender());
226+
}
227+
} catch (IllegalStateException e) {
228+
lock = new ReentrantLock();
229+
this.buffer = new Buffer(ioCallback, lock);
230+
this.buffer.sendOrBuf(buffer, sender());
120231
} catch (RuntimeException e) {
121232
endAsync = false;
122233
throw e;
@@ -177,7 +288,11 @@ public void commit() {
177288
if (!endAsync) {
178289
hse.endExchange();
179290
} else {
180-
sender.close(IoCallback.END_EXCHANGE);
291+
if (null != buffer) {
292+
buffer.sendThroughFinalPart(sender());
293+
} else {
294+
sender().close(IoCallback.END_EXCHANGE);
295+
}
181296
}
182297
markClosed();
183298
}
@@ -246,10 +361,4 @@ private boolean responseStarted() {
246361
return hse.isResponseStarted();
247362
}
248363

249-
private static final IoCallback WRITE_PART_CALLBACK = new DefaultIoCallback() {
250-
@Override
251-
public void onComplete(HttpServerExchange exchange, Sender sender) {
252-
// do not close the exchange
253-
}
254-
};
255364
}

‎src/main/java/act/xio/undertow/UndertowResponseOutput.java‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,6 @@
2020
* #L%
2121
*/
2222

23-
import io.undertow.io.Sender;
2423
import org.osgl.util.Output;
2524

2625
import java.io.OutputStream;
@@ -30,19 +29,19 @@
3029
public class UndertowResponseOutput implements Output {
3130

3231
private UndertowResponse resp;
33-
private Sender sender;
3432

3533
public UndertowResponseOutput(UndertowResponse resp) {
3634
this.resp = resp;
37-
this.sender = resp.sender();
3835
}
3936

4037
@Override
4138
public void open() {
39+
resp.beforeWritingContent();
4240
}
4341

4442
@Override
4543
public void close() {
44+
resp.afterWritingContent();
4645
}
4746

4847
@Override

0 commit comments

Comments
 (0)