Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
爱吃血肠
spring-framework
提交
4e39450d
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,体验更适合开发者的 AI 搜索 >>
提交
4e39450d
编写于
8月 27, 2019
作者:
R
Rossen Stoyanchev
浏览文件
操作
浏览文件
下载
差异文件
Merge branch '5.1.x'
上级
46a37b44
dd22b8fd
变更
3
隐藏空白更改
内联
并排
Showing
3 changed file
with
50 addition
and
27 deletion
+50
-27
spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerReadPublisher.java
...k/http/server/reactive/AbstractListenerReadPublisher.java
+21
-14
spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerWriteFlushProcessor.java
.../server/reactive/AbstractListenerWriteFlushProcessor.java
+20
-12
spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerWriteProcessor.java
.../http/server/reactive/AbstractListenerWriteProcessor.java
+9
-1
未找到文件。
spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerReadPublisher.java
浏览文件 @
4e39450d
/*
* Copyright 2002-201
8
the original author or authors.
* Copyright 2002-201
9
the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
...
...
@@ -283,19 +283,7 @@ public abstract class AbstractListenerReadPublisher<T> implements Publisher<T> {
publisher
.
subscriber
=
subscriber
;
subscriber
.
onSubscribe
(
subscription
);
publisher
.
changeState
(
SUBSCRIBING
,
NO_DEMAND
);
// Now safe to check "beforeDemand" flags, they won't change once in NO_DEMAND
String
logPrefix
=
publisher
.
getLogPrefix
();
if
(
publisher
.
completionBeforeDemand
)
{
rsReadLogger
.
trace
(
logPrefix
+
"Completed before demand"
);
publisher
.
state
.
get
().
onAllDataRead
(
publisher
);
}
Throwable
ex
=
publisher
.
errorBeforeDemand
;
if
(
ex
!=
null
)
{
if
(
rsReadLogger
.
isTraceEnabled
())
{
rsReadLogger
.
trace
(
logPrefix
+
"Completed with error before demand: "
+
ex
);
}
publisher
.
state
.
get
().
onError
(
publisher
,
ex
);
}
handleCompletionOrErrorBeforeDemand
(
publisher
);
}
else
{
throw
new
IllegalStateException
(
"Failed to transition to SUBSCRIBING, "
+
...
...
@@ -306,11 +294,30 @@ public abstract class AbstractListenerReadPublisher<T> implements Publisher<T> {
@Override
<
T
>
void
onAllDataRead
(
AbstractListenerReadPublisher
<
T
>
publisher
)
{
publisher
.
completionBeforeDemand
=
true
;
handleCompletionOrErrorBeforeDemand
(
publisher
);
}
@Override
<
T
>
void
onError
(
AbstractListenerReadPublisher
<
T
>
publisher
,
Throwable
ex
)
{
publisher
.
errorBeforeDemand
=
ex
;
handleCompletionOrErrorBeforeDemand
(
publisher
);
}
private
<
T
>
void
handleCompletionOrErrorBeforeDemand
(
AbstractListenerReadPublisher
<
T
>
publisher
)
{
if
(
publisher
.
state
.
get
().
equals
(
NO_DEMAND
))
{
if
(
publisher
.
completionBeforeDemand
)
{
rsReadLogger
.
trace
(
publisher
.
getLogPrefix
()
+
"Completed before demand"
);
publisher
.
state
.
get
().
onAllDataRead
(
publisher
);
}
Throwable
ex
=
publisher
.
errorBeforeDemand
;
if
(
ex
!=
null
)
{
if
(
rsReadLogger
.
isTraceEnabled
())
{
String
prefix
=
publisher
.
getLogPrefix
();
rsReadLogger
.
trace
(
prefix
+
"Completed with error before demand: "
+
ex
);
}
publisher
.
state
.
get
().
onError
(
publisher
,
ex
);
}
}
}
},
...
...
spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerWriteFlushProcessor.java
浏览文件 @
4e39450d
/*
* Copyright 2002-201
8
the original author or authors.
* Copyright 2002-201
9
the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
...
...
@@ -282,17 +282,7 @@ public abstract class AbstractListenerWriteFlushProcessor<T> implements Processo
}
if
(
processor
.
changeState
(
this
,
REQUESTED
))
{
if
(
processor
.
subscriberCompleted
)
{
if
(
processor
.
isFlushPending
())
{
// Ensure the final flush
processor
.
changeState
(
REQUESTED
,
FLUSHING
);
processor
.
flushIfPossible
();
}
else
if
(
processor
.
changeState
(
REQUESTED
,
COMPLETED
))
{
processor
.
resultPublisher
.
publishComplete
();
}
else
{
processor
.
state
.
get
().
onComplete
(
processor
);
}
handleSubscriberCompleted
(
processor
);
}
else
{
Assert
.
state
(
processor
.
subscription
!=
null
,
"No subscription"
);
...
...
@@ -303,6 +293,24 @@ public abstract class AbstractListenerWriteFlushProcessor<T> implements Processo
@Override
public
<
T
>
void
onComplete
(
AbstractListenerWriteFlushProcessor
<
T
>
processor
)
{
processor
.
subscriberCompleted
=
true
;
// A competing write might have completed very quickly
if
(
processor
.
state
.
get
().
equals
(
State
.
REQUESTED
))
{
handleSubscriberCompleted
(
processor
);
}
}
private
<
T
>
void
handleSubscriberCompleted
(
AbstractListenerWriteFlushProcessor
<
T
>
processor
)
{
if
(
processor
.
isFlushPending
())
{
// Ensure the final flush
processor
.
changeState
(
State
.
REQUESTED
,
State
.
FLUSHING
);
processor
.
flushIfPossible
();
}
else
if
(
processor
.
changeState
(
State
.
REQUESTED
,
State
.
COMPLETED
))
{
processor
.
resultPublisher
.
publishComplete
();
}
else
{
processor
.
state
.
get
().
onComplete
(
processor
);
}
}
},
...
...
spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerWriteProcessor.java
浏览文件 @
4e39450d
/*
* Copyright 2002-201
8
the original author or authors.
* Copyright 2002-201
9
the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
...
...
@@ -376,6 +376,10 @@ public abstract class AbstractListenerWriteProcessor<T> implements Processor<T,
@Override
public
<
T
>
void
onComplete
(
AbstractListenerWriteProcessor
<
T
>
processor
)
{
processor
.
subscriberCompleted
=
true
;
// A competing write might have completed very quickly
if
(
processor
.
state
.
get
().
equals
(
State
.
REQUESTED
))
{
processor
.
changeStateToComplete
(
State
.
REQUESTED
);
}
}
},
...
...
@@ -383,6 +387,10 @@ public abstract class AbstractListenerWriteProcessor<T> implements Processor<T,
@Override
public
<
T
>
void
onComplete
(
AbstractListenerWriteProcessor
<
T
>
processor
)
{
processor
.
subscriberCompleted
=
true
;
// A competing write might have completed very quickly
if
(
processor
.
state
.
get
().
equals
(
State
.
REQUESTED
))
{
processor
.
changeStateToComplete
(
State
.
REQUESTED
);
}
}
},
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录