Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
爱吃血肠
spring-framework
提交
799bb642
S
spring-framework
项目概览
爱吃血肠
/
spring-framework
通知
1
Star
0
Fork
0
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
S
spring-framework
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
提交
799bb642
编写于
7月 08, 2016
作者:
A
Arjen Poutsma
提交者:
GitHub
7月 08, 2016
浏览文件
操作
浏览文件
下载
差异文件
Merge pull request #135 from violetagg/ondataavailable-concurrency
Refactor AbstractRequestBodyPublisher states
上级
d7a43d20
d68232c8
变更
3
隐藏空白更改
内联
并排
Showing
3 changed file
with
74 addition
and
58 deletion
+74
-58
spring-web-reactive/src/main/java/org/springframework/http/server/reactive/AbstractRequestBodyPublisher.java
...rk/http/server/reactive/AbstractRequestBodyPublisher.java
+51
-55
spring-web-reactive/src/main/java/org/springframework/http/server/reactive/ServletHttpHandlerAdapter.java
...ework/http/server/reactive/ServletHttpHandlerAdapter.java
+18
-3
spring-web-reactive/src/main/java/org/springframework/http/server/reactive/UndertowServerHttpRequest.java
...ework/http/server/reactive/UndertowServerHttpRequest.java
+5
-0
未找到文件。
spring-web-reactive/src/main/java/org/springframework/http/server/reactive/AbstractRequestBodyPublisher.java
浏览文件 @
799bb642
...
...
@@ -52,8 +52,6 @@ abstract class AbstractRequestBodyPublisher implements Publisher<DataBuffer> {
private
Subscriber
<?
super
DataBuffer
>
subscriber
;
private
volatile
boolean
dataAvailable
;
@Override
public
void
subscribe
(
Subscriber
<?
super
DataBuffer
>
subscriber
)
{
if
(
this
.
logger
.
isTraceEnabled
())
{
...
...
@@ -101,28 +99,24 @@ abstract class AbstractRequestBodyPublisher implements Publisher<DataBuffer> {
/**
* Reads and publishes data buffers from the input. Continues till either there is no
* more demand, or till there is no more data to be read.
* @return {@code true} if there is more d
ata to be rea
d; {@code false} otherwise
* @return {@code true} if there is more d
eman
d; {@code false} otherwise
*/
private
boolean
readAndPublish
()
{
try
{
while
(
hasDemand
())
{
DataBuffer
dataBuffer
=
read
();
if
(
dataBuffer
!=
null
)
{
BackpressureUtils
.
getAndSub
(
this
.
demand
,
1L
);
this
.
subscriber
.
onNext
(
dataBuffer
);
}
else
{
return
false
;
}
private
boolean
readAndPublish
()
throws
IOException
{
while
(
hasDemand
())
{
DataBuffer
dataBuffer
=
read
();
if
(
dataBuffer
!=
null
)
{
BackpressureUtils
.
getAndSub
(
this
.
demand
,
1L
);
this
.
subscriber
.
onNext
(
dataBuffer
);
}
else
{
return
true
;
}
return
true
;
}
catch
(
IOException
ex
)
{
onError
(
ex
);
return
false
;
}
return
false
;
}
protected
abstract
void
checkOnDataAvailable
();
/**
* Reads a data buffer from the input, if possible. Returns {@code null} if a buffer
* could not be read.
...
...
@@ -182,10 +176,13 @@ abstract class AbstractRequestBodyPublisher implements Publisher<DataBuffer> {
* UNSUBSCRIBED
* |
* v
* DATA_UNAVAILABLE <---> DATA_AVAILABLE
* | |
* v v
* COMPLETED
* NO_DEMAND -------------------> DEMAND
* | ^ ^ |
* | | | |
* | --------- READING <----- |
* | | |
* | v |
* ------------> COMPLETED <---------
* </pre>
* Refer to the individual states for more information.
*/
...
...
@@ -194,16 +191,14 @@ abstract class AbstractRequestBodyPublisher implements Publisher<DataBuffer> {
/**
* The initial unsubscribed state. Will respond to {@link
* #subscribe(AbstractRequestBodyPublisher, Subscriber)} by
* changing state to {@link #
DATA_UNAVAILABLE
}.
* changing state to {@link #
NO_DEMAND
}.
*/
UNSUBSCRIBED
{
@Override
void
subscribe
(
AbstractRequestBodyPublisher
publisher
,
Subscriber
<?
super
DataBuffer
>
subscriber
)
{
Objects
.
requireNonNull
(
subscriber
);
State
newState
=
publisher
.
dataAvailable
?
DATA_AVAILABLE
:
DATA_UNAVAILABLE
;
if
(
publisher
.
changeState
(
this
,
newState
))
{
if
(
publisher
.
changeState
(
this
,
NO_DEMAND
))
{
Subscription
subscription
=
new
RequestBodySubscription
(
publisher
);
publisher
.
subscriber
=
subscriber
;
...
...
@@ -213,54 +208,55 @@ abstract class AbstractRequestBodyPublisher implements Publisher<DataBuffer> {
throw
new
IllegalStateException
(
toString
());
}
}
@Override
void
onDataAvailable
(
AbstractRequestBodyPublisher
publisher
)
{
publisher
.
dataAvailable
=
true
;
}
},
/**
* State that gets entered when there is no data to be read. Responds to {@link
* #request(AbstractRequestBodyPublisher, long)} by increasing the demand, and
* responds to {@link #onDataAvailable(AbstractRequestBodyPublisher)} by
* reading the available data and changing state to {@link #DATA_AVAILABLE} if
* there continues to be more data available after the demand has been satisfied.
* State that gets entered when there is no demand. Responds to {@link
* #request(AbstractRequestBodyPublisher, long)} by increasing the demand,
* changing state to {@link #DEMAND} and will check whether there
* is data available for reading.
*/
DATA_UNAVAILABLE
{
NO_DEMAND
{
@Override
void
request
(
AbstractRequestBodyPublisher
publisher
,
long
n
)
{
if
(
BackpressureUtils
.
checkRequest
(
n
,
publisher
.
subscriber
))
{
BackpressureUtils
.
addAndGet
(
publisher
.
demand
,
n
);
if
(
publisher
.
changeState
(
this
,
DEMAND
))
{
publisher
.
checkOnDataAvailable
();
}
}
}
},
/**
* State that gets entered when there is demand. Responds to
* {@link #onDataAvailable(AbstractRequestBodyPublisher)} by
* reading the available data. The state will be changed to
* {@link #NO_DEMAND} if there is no demand.
*/
DEMAND
{
@Override
void
onDataAvailable
(
AbstractRequestBodyPublisher
publisher
)
{
boolean
dataAvailable
=
publisher
.
readAndPublish
();
if
(
dataAvailable
)
{
publisher
.
changeState
(
this
,
DATA_AVAILABLE
);
if
(
publisher
.
changeState
(
this
,
READING
))
{
try
{
boolean
demandAvailable
=
publisher
.
readAndPublish
();
if
(
demandAvailable
)
{
publisher
.
changeState
(
READING
,
DEMAND
);
publisher
.
checkOnDataAvailable
();
}
else
{
publisher
.
changeState
(
READING
,
NO_DEMAND
);
}
}
catch
(
IOException
ex
)
{
publisher
.
onError
(
ex
);
}
}
}
},
/**
* State that gets entered when there is data to be read. Responds to {@link
* #request(AbstractRequestBodyPublisher, long)} by increasing the demand, and
* by reading the available data and changing state to {@link #DATA_UNAVAILABLE}
* if there is no more data available.
*/
DATA_AVAILABLE
{
READING
{
@Override
void
request
(
AbstractRequestBodyPublisher
publisher
,
long
n
)
{
if
(
BackpressureUtils
.
checkRequest
(
n
,
publisher
.
subscriber
))
{
BackpressureUtils
.
addAndGet
(
publisher
.
demand
,
n
);
boolean
dataAvailable
=
publisher
.
readAndPublish
();
if
(!
dataAvailable
)
{
publisher
.
changeState
(
this
,
DATA_UNAVAILABLE
);
}
}
}
},
/**
* The terminal completed state. Does not respond to any events.
...
...
spring-web-reactive/src/main/java/org/springframework/http/server/reactive/ServletHttpHandlerAdapter.java
浏览文件 @
799bb642
...
...
@@ -159,13 +159,28 @@ public class ServletHttpHandlerAdapter extends HttpServlet {
}
public
void
registerListener
()
throws
IOException
{
this
.
synchronizer
.
getRequest
().
getInputStream
()
.
setReadListener
(
this
.
readListener
);
inputStream
().
setReadListener
(
this
.
readListener
);
}
private
ServletInputStream
inputStream
()
throws
IOException
{
return
this
.
synchronizer
.
getRequest
().
getInputStream
();
}
@Override
protected
void
checkOnDataAvailable
()
{
try
{
if
(!
inputStream
().
isFinished
()
&&
inputStream
().
isReady
())
{
onDataAvailable
();
}
}
catch
(
IOException
ex
)
{
onError
(
ex
);
}
}
@Override
protected
DataBuffer
read
()
throws
IOException
{
ServletInputStream
input
=
this
.
synchronizer
.
getRequest
().
getI
nputStream
();
ServletInputStream
input
=
i
nputStream
();
if
(
input
.
isReady
())
{
int
read
=
input
.
read
(
this
.
buffer
);
if
(
logger
.
isTraceEnabled
())
{
...
...
spring-web-reactive/src/main/java/org/springframework/http/server/reactive/UndertowServerHttpRequest.java
浏览文件 @
799bb642
...
...
@@ -129,6 +129,11 @@ public class UndertowServerHttpRequest extends AbstractServerHttpRequest {
this
.
requestChannel
.
resumeReads
();
}
@Override
protected
void
checkOnDataAvailable
()
{
onDataAvailable
();
}
@Override
protected
DataBuffer
read
()
throws
IOException
{
ByteBuffer
byteBuffer
=
this
.
pooledByteBuffer
.
getBuffer
();
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录