grpc
/
grpc-java
mirror of https://github.com/grpc/grpc-java.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

A couple minor fixes to Channel for bugs exposed while integration testing with Netty transport. 

-------------
Created by MOE: http://code.google.com/p/moe-java
MOE_MIGRATED_REVID=69939586
This commit is contained in:
nathanmittler 2014-06-25 09:18:42 -07:00 committed by Eric Anderson
parent 6933667178
commit f4694f57f9
2 changed files with 16 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
core/src/main/java/com/google/net/stubby/ChannelImpl.java
View File

@ -52,9 +52,13 @@ public final class ChannelImpl extends AbstractService implements Channel {
@Override @Override
protected synchronized void doStop() { protected synchronized void doStop() {
if (activeTransport != null) {
activeTransport.stopAsync(); activeTransport.stopAsync();
activeTransport = null; activeTransport = null;
// The last TransportListener will call notifyStopped(). // The last TransportListener will call notifyStopped().
} else {
notifyStopped();
}
} }
@Override @Override
@ -115,7 +119,7 @@ public final class ChannelImpl extends AbstractService implements Channel {
private class CallImpl<ReqT, RespT> extends Call<ReqT, RespT> { private class CallImpl<ReqT, RespT> extends Call<ReqT, RespT> {
private final MethodDescriptor<ReqT, RespT> method; private final MethodDescriptor<ReqT, RespT> method;
private final SerializingExecutor executor; private final SerializingExecutor callExecutor;
// TODO(user): Consider moving flow control notification/management to Call itself. // TODO(user): Consider moving flow control notification/management to Call itself.
private final Collection<SettableFuture<Void>> inProcessFutures private final Collection<SettableFuture<Void>> inProcessFutures
= Collections.synchronizedSet(new HashSet<SettableFuture<Void>>()); = Collections.synchronizedSet(new HashSet<SettableFuture<Void>>());
@ -123,7 +127,7 @@ public final class ChannelImpl extends AbstractService implements Channel {
public CallImpl(MethodDescriptor<ReqT, RespT> method, SerializingExecutor executor) { public CallImpl(MethodDescriptor<ReqT, RespT> method, SerializingExecutor executor) {
this.method = method; this.method = method;
this.executor = executor; this.callExecutor = executor;
} }
@Override @Override
@ -134,9 +138,12 @@ public final class ChannelImpl extends AbstractService implements Channel {
@Override @Override
public void cancel() { public void cancel() {
Preconditions.checkState(stream != null, "Not started"); // Cancel is called in exception handling cases, so it may be the case that the
// stream was never successfully created.
if (stream != null) {
stream.cancel(); stream.cancel();
} }
}
@Override @Override
public void halfClose() { public void halfClose() {
@ -200,7 +207,7 @@ public final class ChannelImpl extends AbstractService implements Channel {
private ListenableFuture<Void> dispatchCallable( private ListenableFuture<Void> dispatchCallable(
final Callable<ListenableFuture<Void>> callable) { final Callable<ListenableFuture<Void>> callable) {
final SettableFuture<Void> ours = SettableFuture.create(); final SettableFuture<Void> ours = SettableFuture.create();
executor.execute(new Runnable() { callExecutor.execute(new Runnable() {
@Override @Override
public void run() { public void run() {
try { try {
@ -252,7 +259,7 @@ public final class ChannelImpl extends AbstractService implements Channel {
future.cancel(false); future.cancel(false);
} }
inProcessFutures.clear(); inProcessFutures.clear();
executor.execute(new Runnable() { callExecutor.execute(new Runnable() {
@Override @Override
public void run() { public void run() {
observer.onClose(status); observer.onClose(status);

1
stub/src/main/java/com/google/net/stubby/stub/Calls.java
View File

@ -18,7 +18,6 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future; import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.LinkedBlockingQueue;
/** /**
* Utility functions for processing different call idioms. We have one-to-one correspondence * Utility functions for processing different call idioms. We have one-to-one correspondence
* between utilities in this class and the potential signatures in a generated stub class so * between utilities in this class and the potential signatures in a generated stub class so