Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
thythm
ribbon
提交
e9035c99
R
ribbon
项目概览
thythm
/
ribbon
与 Fork 源项目一致
从无法访问的项目Fork
通知
3
Star
0
Fork
0
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
R
ribbon
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
提交
e9035c99
编写于
10月 05, 2013
作者:
A
Allen Wang
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
Added streaming support for Observable.
上级
4b86cd5e
变更
2
隐藏空白更改
内联
并排
Showing
2 changed file
with
100 addition
and
4 deletion
+100
-4
ribbon-core/src/main/java/com/netflix/client/ObservableAsyncClient.java
...c/main/java/com/netflix/client/ObservableAsyncClient.java
+79
-3
ribbon-httpasyncclient/src/test/java/com/netflix/httpasyncclient/HttpAsyncClienTest.java
.../java/com/netflix/httpasyncclient/HttpAsyncClienTest.java
+21
-1
未找到文件。
ribbon-core/src/main/java/com/netflix/client/ObservableAsyncClient.java
浏览文件 @
e9035c99
package
com.netflix.client
;
import
java.util.concurrent.Future
;
import
rx.Observable
;
import
rx.Observer
;
import
rx.Subscription
;
...
...
@@ -8,11 +10,29 @@ import rx.Observable.OnSubscribeFunc;
import
rx.subscriptions.CompositeSubscription
;
import
rx.subscriptions.Subscriptions
;
public
class
ObservableAsyncClient
<
T
extends
ClientRequest
,
S
extends
ResponseWithTypedEntity
>
{
public
class
ObservableAsyncClient
<
T
extends
ClientRequest
,
S
extends
IResponse
,
U
>
{
public
static
class
StreamEvent
<
U
extends
IResponse
,
E
>
{
private
volatile
U
response
;
private
volatile
E
event
;
private
final
AsyncClient
<
T
,
S
,
?>
client
;
public
StreamEvent
(
U
response
,
E
event
)
{
super
();
this
.
response
=
response
;
this
.
event
=
event
;
}
public
final
U
getResponse
()
{
return
response
;
}
public
final
E
getEvent
()
{
return
event
;
}
}
private
final
AsyncClient
<
T
,
S
,
U
>
client
;
public
ObservableAsyncClient
(
AsyncClient
<
T
,
S
,
?
>
client
)
{
public
ObservableAsyncClient
(
AsyncClient
<
T
,
S
,
U
>
client
)
{
this
.
client
=
client
;
}
...
...
@@ -56,4 +76,60 @@ public class ObservableAsyncClient<T extends ClientRequest, S extends ResponseWi
}
});
}
public
<
E
>
Observable
<
StreamEvent
<
S
,
E
>>
stream
(
final
T
request
,
final
StreamDecoder
<
E
,
U
>
decoder
)
{
final
OnSubscribeFunc
<
StreamEvent
<
S
,
E
>>
onSubscribeFunc
=
new
OnSubscribeFunc
<
StreamEvent
<
S
,
E
>>()
{
@Override
public
Subscription
onSubscribe
(
final
Observer
<?
super
StreamEvent
<
S
,
E
>>
observer
)
{
final
CompositeSubscription
parentSubscription
=
new
CompositeSubscription
();
try
{
Future
<?>
future
=
client
.
execute
(
request
,
(
StreamDecoder
<
E
,
U
>)
decoder
,
new
ResponseCallback
<
S
,
E
>()
{
private
volatile
S
response
;
@Override
public
void
completed
(
S
response
)
{
observer
.
onCompleted
();
}
@Override
public
void
failed
(
Throwable
e
)
{
observer
.
onError
(
e
);
}
@Override
public
void
cancelled
()
{
observer
.
onError
(
new
IllegalStateException
(
"operation cancelled"
));
}
@Override
public
void
responseReceived
(
S
response
)
{
this
.
response
=
response
;
}
@Override
public
void
contentReceived
(
E
content
)
{
StreamEvent
<
S
,
E
>
e
=
new
StreamEvent
<
S
,
E
>(
this
.
response
,
content
);
observer
.
onNext
(
e
);
}
}
);
parentSubscription
.
add
(
Subscriptions
.
from
(
future
));
}
catch
(
ClientException
e
)
{
// TODO Auto-generated catch block
e
.
printStackTrace
();
}
return
parentSubscription
;
}
};
return
Observable
.
create
(
new
OnSubscribeFunc
<
StreamEvent
<
S
,
E
>>()
{
@Override
public
Subscription
onSubscribe
(
final
Observer
<?
super
StreamEvent
<
S
,
E
>>
observer
)
{
return
onSubscribeFunc
.
onSubscribe
(
observer
);
}
});
}
}
ribbon-httpasyncclient/src/test/java/com/netflix/httpasyncclient/HttpAsyncClienTest.java
浏览文件 @
e9035c99
...
...
@@ -32,6 +32,7 @@ import com.netflix.client.AsyncLoadBalancingClient;
import
com.netflix.client.ClientException
;
import
com.netflix.client.FullResponseCallback
;
import
com.netflix.client.ObservableAsyncClient
;
import
com.netflix.client.ObservableAsyncClient.StreamEvent
;
import
com.netflix.client.ResponseCallback
;
import
com.netflix.client.StreamDecoder
;
import
com.netflix.client.config.CommonClientConfigKey
;
...
...
@@ -187,7 +188,7 @@ public class HttpAsyncClienTest {
public
void
testObservable
()
throws
Exception
{
URI
uri
=
new
URI
(
SERVICE_URI
+
"testNetty/person"
);
HttpRequest
request
=
HttpRequest
.
newBuilder
().
uri
(
uri
).
build
();
ObservableAsyncClient
<
HttpRequest
,
HttpResponse
>
observableClient
=
new
ObservableAsyncClient
<
HttpRequest
,
HttpResponse
>(
client
);
ObservableAsyncClient
<
HttpRequest
,
HttpResponse
,
ByteBuffer
>
observableClient
=
new
ObservableAsyncClient
<
HttpRequest
,
HttpResponse
,
ByteBuffer
>(
client
);
final
List
<
Person
>
result
=
Lists
.
newArrayList
();
observableClient
.
execute
(
request
).
toBlockingObservable
().
forEach
(
new
Action1
<
HttpResponse
>()
{
@Override
...
...
@@ -532,6 +533,25 @@ public class HttpAsyncClienTest {
assertEquals
(
EmbeddedResources
.
streamContent
,
results
);
}
@Test
public
void
testStreamObservable
()
throws
Exception
{
HttpRequest
request
=
HttpRequest
.
newBuilder
().
uri
(
SERVICE_URI
+
"testNetty/stream"
).
build
();
final
List
<
String
>
results
=
Lists
.
newArrayList
();
ObservableAsyncClient
<
HttpRequest
,
HttpResponse
,
ByteBuffer
>
observableClient
=
new
ObservableAsyncClient
<
HttpRequest
,
HttpResponse
,
ByteBuffer
>(
client
);
observableClient
.
stream
(
request
,
new
SSEDecoder
())
.
toBlockingObservable
()
.
forEach
(
new
Action1
<
StreamEvent
<
HttpResponse
,
List
<
String
>>>()
{
@Override
public
void
call
(
final
StreamEvent
<
HttpResponse
,
List
<
String
>>
t1
)
{
results
.
addAll
(
t1
.
getEvent
());
}
});
assertEquals
(
EmbeddedResources
.
streamContent
,
results
);
}
@Test
public
void
testStreamWithLoadBalancer
()
throws
Exception
{
AsyncLoadBalancingClient
<
HttpRequest
,
HttpResponse
,
ByteBuffer
>
loadBalancingClient
=
new
AsyncLoadBalancingClient
<
HttpRequest
,
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录