Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
wd1105040417
retrofit
提交
04f62bea
R
retrofit
项目概览
wd1105040417
/
retrofit
与 Fork 源项目一致
从无法访问的项目Fork
通知
2
Star
0
Fork
0
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
R
retrofit
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
前往新版Gitcode,体验更适合开发者的 AI 搜索 >>
提交
04f62bea
编写于
5月 08, 2014
作者:
J
Jake Wharton
浏览文件
操作
浏览文件
下载
差异文件
Merge branch 'chrisjenx-fix/rxjava-0-18'
上级
b73a44d8
2602afdf
变更
5
隐藏空白更改
内联
并排
Showing
5 changed file
with
254 addition
and
55 deletion
+254
-55
pom.xml
pom.xml
+1
-1
retrofit-mock/src/main/java/retrofit/MockRestAdapter.java
retrofit-mock/src/main/java/retrofit/MockRestAdapter.java
+21
-16
retrofit/src/main/java/retrofit/RestAdapter.java
retrofit/src/main/java/retrofit/RestAdapter.java
+0
-38
retrofit/src/main/java/retrofit/RxSupport.java
retrofit/src/main/java/retrofit/RxSupport.java
+58
-0
retrofit/src/test/java/retrofit/RxSupportTest.java
retrofit/src/test/java/retrofit/RxSupportTest.java
+174
-0
未找到文件。
pom.xml
浏览文件 @
04f62bea
...
...
@@ -51,7 +51,7 @@
<android.platform>
16
</android.platform>
<gson.version>
2.2.4
</gson.version>
<okhttp.version>
1.3.0
</okhttp.version>
<rxjava.version>
0.1
7.1
</rxjava.version>
<rxjava.version>
0.1
8.3
</rxjava.version>
<appengine.version>
1.8.9
</appengine.version>
<!-- Converter Dependencies -->
...
...
retrofit-mock/src/main/java/retrofit/MockRestAdapter.java
浏览文件 @
04f62bea
...
...
@@ -8,13 +8,12 @@ import java.lang.reflect.Method;
import
java.lang.reflect.Proxy
;
import
java.util.Map
;
import
java.util.Random
;
import
java.util.concurrent.Executor
;
import
java.util.concurrent.TimeUnit
;
import
retrofit.client.Request
;
import
retrofit.client.Response
;
import
rx.Observable
;
import
rx.Scheduler
;
import
rx.Subscriber
;
import
rx.schedulers.Schedulers
;
import
static
retrofit
.
RestAdapter
.
LogLevel
;
import
static
retrofit
.
RetrofitError
.
unexpectedError
;
...
...
@@ -526,30 +525,36 @@ public final class MockRestAdapter {
/** Indirection to avoid VerifyError if RxJava isn't present. */
private
static
class
MockRxSupport
{
private
final
Scheduler
schedule
r
;
private
final
Executor
httpExecuto
r
;
private
final
ErrorHandler
errorHandler
;
MockRxSupport
(
RestAdapter
restAdapter
)
{
scheduler
=
Schedulers
.
executor
(
restAdapter
.
httpExecutor
)
;
httpExecutor
=
restAdapter
.
httpExecutor
;
errorHandler
=
restAdapter
.
errorHandler
;
}
Observable
createMockObservable
(
final
MockHandler
mockHandler
,
final
RestMethodInfo
methodInfo
,
final
RequestInterceptor
interceptor
,
final
Object
[]
args
)
{
return
Observable
.
create
(
new
Observable
.
OnSubscribe
<
Object
>()
{
@Override
public
void
call
(
Subscriber
<?
super
Object
>
subscriber
)
{
try
{
Observable
observable
=
(
Observable
)
mockHandler
.
invokeSync
(
methodInfo
,
interceptor
,
args
);
//noinspection unchecked
observable
.
subscribe
(
subscriber
);
}
catch
(
RetrofitError
e
)
{
subscriber
.
onError
(
errorHandler
.
handleError
(
e
));
}
catch
(
Throwable
e
)
{
subscriber
.
onError
(
e
);
}
@Override
public
void
call
(
final
Subscriber
<?
super
Object
>
subscriber
)
{
if
(
subscriber
.
isUnsubscribed
())
return
;
httpExecutor
.
execute
(
new
Runnable
()
{
@Override
public
void
run
()
{
try
{
if
(
subscriber
.
isUnsubscribed
())
return
;
Observable
observable
=
(
Observable
)
mockHandler
.
invokeSync
(
methodInfo
,
interceptor
,
args
);
//noinspection unchecked
observable
.
subscribe
(
subscriber
);
}
catch
(
RetrofitError
e
)
{
subscriber
.
onError
(
errorHandler
.
handleError
(
e
));
}
catch
(
Throwable
e
)
{
subscriber
.
onError
(
e
);
}
}
});
}
})
.
subscribeOn
(
scheduler
)
;
});
}
}
}
retrofit/src/main/java/retrofit/RestAdapter.java
浏览文件 @
04f62bea
...
...
@@ -38,10 +38,6 @@ import retrofit.mime.MimeUtil;
import
retrofit.mime.TypedByteArray
;
import
retrofit.mime.TypedInput
;
import
retrofit.mime.TypedOutput
;
import
rx.Observable
;
import
rx.Scheduler
;
import
rx.Subscriber
;
import
rx.schedulers.Schedulers
;
/**
* Adapts a Java interface to a REST API.
...
...
@@ -222,40 +218,6 @@ public class RestAdapter {
}
}
/** Indirection to avoid VerifyError if RxJava isn't present. */
private
static
final
class
RxSupport
{
private
final
Scheduler
scheduler
;
private
final
ErrorHandler
errorHandler
;
RxSupport
(
Executor
executor
,
ErrorHandler
errorHandler
)
{
this
.
scheduler
=
Schedulers
.
executor
(
executor
);
this
.
errorHandler
=
errorHandler
;
}
Observable
createRequestObservable
(
final
Callable
<
ResponseWrapper
>
request
)
{
return
Observable
.
create
(
new
Observable
.
OnSubscribe
<
Object
>()
{
@Override
public
void
call
(
Subscriber
<?
super
Object
>
subscriber
)
{
if
(
subscriber
.
isUnsubscribed
())
{
return
;
}
try
{
ResponseWrapper
wrapper
=
request
.
call
();
if
(
subscriber
.
isUnsubscribed
())
{
return
;
}
subscriber
.
onNext
(
wrapper
.
responseBody
);
subscriber
.
onCompleted
();
}
catch
(
RetrofitError
e
)
{
subscriber
.
onError
(
errorHandler
.
handleError
(
e
));
}
catch
(
Exception
e
)
{
// This is from the Callable. It shouldn't actually throw.
throw
new
RuntimeException
(
e
);
}
}
}).
subscribeOn
(
scheduler
);
}
}
private
class
RestHandler
implements
InvocationHandler
{
private
final
Map
<
Method
,
RestMethodInfo
>
methodDetailsCache
;
...
...
retrofit/src/main/java/retrofit/RxSupport.java
0 → 100644
浏览文件 @
04f62bea
package
retrofit
;
import
java.util.concurrent.Callable
;
import
java.util.concurrent.Executor
;
import
java.util.concurrent.FutureTask
;
import
rx.Observable
;
import
rx.Subscriber
;
import
rx.subscriptions.Subscriptions
;
/**
* Utilities for supporting RxJava Observables.
* <p>
* RxJava might not be on the available to use. Check {@link Platform#HAS_RX_JAVA} before calling.
*/
final
class
RxSupport
{
private
final
Executor
executor
;
private
final
ErrorHandler
errorHandler
;
RxSupport
(
Executor
executor
,
ErrorHandler
errorHandler
)
{
this
.
executor
=
executor
;
this
.
errorHandler
=
errorHandler
;
}
Observable
createRequestObservable
(
final
Callable
<
ResponseWrapper
>
request
)
{
return
Observable
.
create
(
new
Observable
.
OnSubscribe
<
Object
>()
{
@Override
public
void
call
(
Subscriber
<?
super
Object
>
subscriber
)
{
if
(
subscriber
.
isUnsubscribed
())
{
return
;
}
FutureTask
<
Void
>
task
=
new
FutureTask
<
Void
>(
getRunnable
(
subscriber
,
request
),
null
);
// Subscribe to the future task of the network call allowing unsubscription.
subscriber
.
add
(
Subscriptions
.
from
(
task
));
executor
.
execute
(
task
);
}
});
}
private
Runnable
getRunnable
(
final
Subscriber
<?
super
Object
>
subscriber
,
final
Callable
<
ResponseWrapper
>
request
)
{
return
new
Runnable
()
{
@Override
public
void
run
()
{
try
{
if
(
subscriber
.
isUnsubscribed
())
{
return
;
}
ResponseWrapper
wrapper
=
request
.
call
();
subscriber
.
onNext
(
wrapper
.
responseBody
);
subscriber
.
onCompleted
();
}
catch
(
RetrofitError
e
)
{
subscriber
.
onError
(
errorHandler
.
handleError
(
e
));
}
catch
(
Exception
e
)
{
// This is from the Callable. It shouldn't actually throw.
throw
new
RuntimeException
(
e
);
}
}
};
}
}
retrofit/src/test/java/retrofit/RxSupportTest.java
0 → 100644
浏览文件 @
04f62bea
package
retrofit
;
import
java.util.ArrayDeque
;
import
java.util.Collections
;
import
java.util.Deque
;
import
java.util.Iterator
;
import
java.util.concurrent.Callable
;
import
java.util.concurrent.Executor
;
import
java.util.concurrent.TimeUnit
;
import
org.junit.Before
;
import
org.junit.Test
;
import
org.mockito.Mock
;
import
org.mockito.MockitoAnnotations
;
import
retrofit.client.Header
;
import
retrofit.client.Response
;
import
retrofit.mime.TypedInput
;
import
rx.Observer
;
import
rx.Subscription
;
import
rx.schedulers.Schedulers
;
import
rx.schedulers.TestScheduler
;
import
static
org
.
mockito
.
Matchers
.
any
;
import
static
org
.
mockito
.
Mockito
.
mock
;
import
static
org
.
mockito
.
Mockito
.
never
;
import
static
org
.
mockito
.
Mockito
.
spy
;
import
static
org
.
mockito
.
Mockito
.
times
;
import
static
org
.
mockito
.
Mockito
.
verify
;
public
class
RxSupportTest
{
private
Object
response
;
private
ResponseWrapper
responseWrapper
;
private
Callable
<
ResponseWrapper
>
callable
=
spy
(
new
Callable
<
ResponseWrapper
>()
{
@Override
public
ResponseWrapper
call
()
throws
Exception
{
return
responseWrapper
;
}
});
private
QueuedSynchronousExecutor
executor
;
private
RxSupport
rxSupport
;
@Mock
Observer
<
Object
>
subscriber
;
@Before
public
void
setUp
()
{
MockitoAnnotations
.
initMocks
(
this
);
response
=
new
Object
();
responseWrapper
=
new
ResponseWrapper
(
new
Response
(
"http://example.com"
,
200
,
"Success"
,
Collections
.<
Header
>
emptyList
(),
mock
(
TypedInput
.
class
)
),
response
);
executor
=
spy
(
new
QueuedSynchronousExecutor
());
rxSupport
=
new
RxSupport
(
executor
,
ErrorHandler
.
DEFAULT
);
}
@Test
public
void
testObservableCallsOnNextOnHttpExecutor
()
{
rxSupport
.
createRequestObservable
(
callable
).
subscribe
(
subscriber
);
executor
.
executeNextInQueue
();
verify
(
subscriber
,
times
(
1
)).
onNext
(
response
);
}
@Test
public
void
testObservableCallsOnNextOnHttpExecutorWithSubscriber
()
{
TestScheduler
test
=
Schedulers
.
test
();
rxSupport
.
createRequestObservable
(
callable
).
subscribeOn
(
test
).
subscribe
(
subscriber
);
// Subscription is handled via the Scheduler.
test
.
triggerActions
();
// This will only execute up to the executor in OnSubscribe.
verify
(
subscriber
,
never
()).
onNext
(
any
());
// Upon continuing the executor we then run the retrofit request.
executor
.
executeNextInQueue
();
verify
(
subscriber
,
times
(
1
)).
onNext
(
response
);
}
@Test
public
void
testObservableUnSubscribesDoesNotExecuteCallable
()
throws
Exception
{
Subscription
subscription
=
rxSupport
.
createRequestObservable
(
callable
).
subscribe
(
subscriber
);
verify
(
subscriber
,
never
()).
onNext
(
any
());
// UnSubscribe here should cancel the queued runnable.
subscription
.
unsubscribe
();
executor
.
executeNextInQueue
();
verify
(
callable
,
never
()).
call
();
verify
(
subscriber
,
never
()).
onNext
(
response
);
}
@Test
public
void
testObservableCallsOperatorsOffHttpExecutor
()
{
TestScheduler
test
=
Schedulers
.
test
();
rxSupport
.
createRequestObservable
(
callable
)
.
delaySubscription
(
1000
,
TimeUnit
.
MILLISECONDS
,
test
)
.
subscribe
(
subscriber
);
verify
(
subscriber
,
never
()).
onNext
(
any
());
test
.
advanceTimeBy
(
1000
,
TimeUnit
.
MILLISECONDS
);
// Upon continuing the executor we then run the retrofit request.
executor
.
executeNextInQueue
();
verify
(
subscriber
,
times
(
1
)).
onNext
(
response
);
}
@Test
public
void
testObservableDoesNotLockExecutor
()
{
TestScheduler
test
=
Schedulers
.
test
();
rxSupport
.
createRequestObservable
(
callable
)
.
delay
(
1000
,
TimeUnit
.
MILLISECONDS
,
test
)
.
subscribe
(
subscriber
);
rxSupport
.
createRequestObservable
(
callable
)
.
delay
(
2000
,
TimeUnit
.
MILLISECONDS
,
test
)
.
subscribe
(
subscriber
);
// Nothing fired yet
verify
(
subscriber
,
never
()).
onNext
(
any
());
// Subscriptions should of been queued up and executed even tho we delayed on the Subscriber.
executor
.
executeNextInQueue
();
executor
.
executeNextInQueue
();
verify
(
subscriber
,
never
()).
onNext
(
response
);
test
.
advanceTimeBy
(
1000
,
TimeUnit
.
MILLISECONDS
);
verify
(
subscriber
,
times
(
1
)).
onNext
(
response
);
test
.
advanceTimeBy
(
1000
,
TimeUnit
.
MILLISECONDS
);
verify
(
subscriber
,
times
(
2
)).
onNext
(
response
);
}
@Test
public
void
testObservableRespectsObserveOn
()
throws
Exception
{
TestScheduler
observe
=
Schedulers
.
test
();
rxSupport
.
createRequestObservable
(
callable
)
.
observeOn
(
observe
)
.
subscribe
(
subscriber
);
verify
(
subscriber
,
never
()).
onNext
(
any
());
executor
.
executeNextInQueue
();
// Should have no response yet, but callback should of been executed.
verify
(
subscriber
,
never
()).
onNext
(
any
());
verify
(
callable
,
times
(
1
)).
call
();
// Forward the Observable Scheduler
observe
.
triggerActions
();
verify
(
subscriber
,
times
(
1
)).
onNext
(
response
);
}
/**
* Test Executor to iterate through Executions to aid in checking
* that the Observable implementation is correct.
*/
static
class
QueuedSynchronousExecutor
implements
Executor
{
Deque
<
Runnable
>
runnableQueue
=
new
ArrayDeque
<
Runnable
>();
@Override
public
void
execute
(
Runnable
runnable
)
{
runnableQueue
.
add
(
runnable
);
}
/**
* Will throw exception if you are expecting something to be added to the Executor
* and it hasn't.
*/
void
executeNextInQueue
()
{
runnableQueue
.
removeFirst
().
run
();
}
/**
* Executes any queued executions on the executor.
*/
void
executeAll
()
{
Iterator
<
Runnable
>
iterator
=
runnableQueue
.
iterator
();
while
(
iterator
.
hasNext
())
{
Runnable
next
=
iterator
.
next
();
next
.
run
();
iterator
.
remove
();
}
}
}
}
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录