1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18 package org.apache.hadoop.hbase.ipc;
19
20 import java.io.IOException;
21 import java.net.ConnectException;
22 import java.net.InetSocketAddress;
23 import java.net.SocketException;
24 import java.nio.ByteBuffer;
25 import java.security.PrivilegedExceptionAction;
26 import java.util.ArrayList;
27 import java.util.HashMap;
28 import java.util.Iterator;
29 import java.util.List;
30 import java.util.Map;
31 import java.util.Random;
32 import java.util.concurrent.TimeUnit;
33
34 import javax.security.sasl.SaslException;
35
36 import org.apache.commons.logging.Log;
37 import org.apache.commons.logging.LogFactory;
38 import org.apache.hadoop.hbase.HConstants;
39 import org.apache.hadoop.hbase.classification.InterfaceAudience;
40 import org.apache.hadoop.hbase.client.MetricsConnection;
41 import org.apache.hadoop.hbase.exceptions.ConnectionClosingException;
42 import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
43 import org.apache.hadoop.hbase.protobuf.generated.AuthenticationProtos;
44 import org.apache.hadoop.hbase.protobuf.generated.AuthenticationProtos.TokenIdentifier.Kind;
45 import org.apache.hadoop.hbase.protobuf.generated.RPCProtos;
46 import org.apache.hadoop.hbase.protobuf.generated.TracingProtos;
47 import org.apache.hadoop.hbase.security.AuthMethod;
48 import org.apache.hadoop.hbase.security.SaslClientHandler;
49 import org.apache.hadoop.hbase.security.SaslUtil;
50 import org.apache.hadoop.hbase.security.SecurityInfo;
51 import org.apache.hadoop.hbase.security.User;
52 import org.apache.hadoop.hbase.security.token.AuthenticationTokenSelector;
53 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
54 import org.apache.hadoop.io.Text;
55 import org.apache.hadoop.ipc.RemoteException;
56 import org.apache.hadoop.security.SecurityUtil;
57 import org.apache.hadoop.security.UserGroupInformation;
58 import org.apache.hadoop.security.token.Token;
59 import org.apache.hadoop.security.token.TokenIdentifier;
60 import org.apache.hadoop.security.token.TokenSelector;
61 import org.apache.htrace.Span;
62 import org.apache.htrace.Trace;
63
64 import com.google.protobuf.Descriptors;
65 import com.google.protobuf.Message;
66 import com.google.protobuf.RpcCallback;
67
68 import io.netty.bootstrap.Bootstrap;
69 import io.netty.buffer.ByteBuf;
70 import io.netty.buffer.ByteBufOutputStream;
71 import io.netty.channel.Channel;
72 import io.netty.channel.ChannelFuture;
73 import io.netty.channel.ChannelFutureListener;
74 import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
75 import io.netty.util.Timeout;
76 import io.netty.util.TimerTask;
77 import io.netty.util.concurrent.GenericFutureListener;
78 import io.netty.util.concurrent.Promise;
79
80
81
82
83 @InterfaceAudience.Private
84 public class AsyncRpcChannel {
85 private static final Log LOG = LogFactory.getLog(AsyncRpcChannel.class.getName());
86
87 private static final int MAX_SASL_RETRIES = 5;
88
89 protected final static Map<Kind, TokenSelector<? extends TokenIdentifier>> TOKEN_HANDDLERS
90 = new HashMap<>();
91
92 static {
93 TOKEN_HANDDLERS.put(AuthenticationProtos.TokenIdentifier.Kind.HBASE_AUTH_TOKEN,
94 new AuthenticationTokenSelector());
95 }
96
97 final AsyncRpcClient client;
98
99
100
101 private Channel channel;
102
103 String name;
104 final User ticket;
105 final String serviceName;
106 final InetSocketAddress address;
107
108 private int ioFailureCounter = 0;
109 private int connectFailureCounter = 0;
110
111 boolean useSasl;
112 AuthMethod authMethod;
113 private int reloginMaxBackoff;
114 private Token<? extends TokenIdentifier> token;
115 private String serverPrincipal;
116
117
118 private final Map<Integer, AsyncCall> pendingCalls = new HashMap<Integer, AsyncCall>();
119 private boolean connected = false;
120 private boolean closed = false;
121
122 private Timeout cleanupTimer;
123
124 private final TimerTask timeoutTask = new TimerTask() {
125 @Override
126 public void run(Timeout timeout) throws Exception {
127 cleanupCalls();
128 }
129 };
130
131
132
133
134
135
136
137
138
139 public AsyncRpcChannel(Bootstrap bootstrap, final AsyncRpcClient client, User ticket,
140 String serviceName, InetSocketAddress address) {
141 this.client = client;
142
143 this.ticket = ticket;
144 this.serviceName = serviceName;
145 this.address = address;
146
147 this.channel = connect(bootstrap).channel();
148
149 name = ("IPC Client (" + channel.hashCode() + ") to " + address.toString()
150 + ((ticket == null) ? " from unknown user" : (" from " + ticket.getName())));
151 }
152
153
154
155
156
157
158 private ChannelFuture connect(final Bootstrap bootstrap) {
159 return bootstrap.remoteAddress(address).connect()
160 .addListener(new GenericFutureListener<ChannelFuture>() {
161 @Override
162 public void operationComplete(final ChannelFuture f) throws Exception {
163 if (!f.isSuccess()) {
164 if (f.cause() instanceof SocketException) {
165 retryOrClose(bootstrap, connectFailureCounter++, f.cause());
166 } else {
167 retryOrClose(bootstrap, ioFailureCounter++, f.cause());
168 }
169 return;
170 }
171 channel = f.channel();
172
173 setupAuthorization();
174
175 ByteBuf b = channel.alloc().directBuffer(6);
176 createPreamble(b, authMethod);
177 channel.writeAndFlush(b).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
178 if (useSasl) {
179 UserGroupInformation ticket = AsyncRpcChannel.this.ticket.getUGI();
180 if (authMethod == AuthMethod.KERBEROS) {
181 if (ticket != null && ticket.getRealUser() != null) {
182 ticket = ticket.getRealUser();
183 }
184 }
185 SaslClientHandler saslHandler;
186 if (ticket == null) {
187 throw new FatalConnectionException("ticket/user is null");
188 }
189 final UserGroupInformation realTicket = ticket;
190 saslHandler = ticket.doAs(new PrivilegedExceptionAction<SaslClientHandler>() {
191 @Override
192 public SaslClientHandler run() throws IOException {
193 return getSaslHandler(realTicket, bootstrap);
194 }
195 });
196 if (saslHandler != null) {
197
198 channel.pipeline().addFirst(saslHandler);
199 } else {
200
201 authMethod = AuthMethod.SIMPLE;
202 useSasl = false;
203 }
204 } else {
205 startHBaseConnection(f.channel());
206 }
207 }
208 });
209 }
210
211
212
213
214
215 private void startHBaseConnection(Channel ch) {
216 ch.pipeline().addLast("frameDecoder",
217 new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
218 ch.pipeline().addLast(new AsyncServerResponseHandler(this));
219 try {
220 writeChannelHeader(ch).addListener(new GenericFutureListener<ChannelFuture>() {
221 @Override
222 public void operationComplete(ChannelFuture future) throws Exception {
223 if (!future.isSuccess()) {
224 close(future.cause());
225 return;
226 }
227 List<AsyncCall> callsToWrite;
228 synchronized (pendingCalls) {
229 connected = true;
230 callsToWrite = new ArrayList<AsyncCall>(pendingCalls.values());
231 }
232 for (AsyncCall call : callsToWrite) {
233 writeRequest(call);
234 }
235 }
236 });
237 } catch (IOException e) {
238 close(e);
239 }
240 }
241
242
243
244
245
246 private void startConnectionWithEncryption(Channel ch) {
247
248
249
250
251 ch.pipeline().addFirst("beforeUnwrapDecoder",
252 new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 0));
253 ch.pipeline().addLast("afterUnwrapDecoder",
254 new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
255 ch.pipeline().addLast(new AsyncServerResponseHandler(this));
256 List<AsyncCall> callsToWrite;
257 synchronized (pendingCalls) {
258 connected = true;
259 callsToWrite = new ArrayList<AsyncCall>(pendingCalls.values());
260 }
261 for (AsyncCall call : callsToWrite) {
262 writeRequest(call);
263 }
264 }
265
266
267
268
269
270
271
272 private SaslClientHandler getSaslHandler(final UserGroupInformation realTicket,
273 final Bootstrap bootstrap) throws IOException {
274 return new SaslClientHandler(realTicket, authMethod, token, serverPrincipal,
275 client.fallbackAllowed,
276 client.conf.get("hbase.rpc.protection",
277 SaslUtil.QualityOfProtection.AUTHENTICATION.name().toLowerCase()),
278 getChannelHeaderBytes(authMethod),
279 new SaslClientHandler.SaslExceptionHandler() {
280 @Override
281 public void handle(int retryCount, Random random, Throwable cause) {
282 try {
283
284 handleSaslConnectionFailure(retryCount, cause, realTicket);
285
286
287 client.newTimeout(new TimerTask() {
288 @Override
289 public void run(Timeout timeout) throws Exception {
290 connect(bootstrap);
291 }
292 }, random.nextInt(reloginMaxBackoff) + 1, TimeUnit.MILLISECONDS);
293 } catch (IOException | InterruptedException e) {
294 close(e);
295 }
296 }
297 }, new SaslClientHandler.SaslSuccessfulConnectHandler() {
298 @Override
299 public void onSuccess(Channel channel) {
300 startHBaseConnection(channel);
301 }
302
303 @Override
304 public void onSaslProtectionSucess(Channel channel) {
305 startConnectionWithEncryption(channel);
306 }
307 });
308 }
309
310
311
312
313
314
315
316
317 private void retryOrClose(final Bootstrap bootstrap, int connectCounter, Throwable e) {
318 if (connectCounter < client.maxRetries) {
319 client.newTimeout(new TimerTask() {
320 @Override public void run(Timeout timeout) throws Exception {
321 connect(bootstrap);
322 }
323 }, client.failureSleep, TimeUnit.MILLISECONDS);
324 } else {
325 client.failedServers.addToFailedServers(address);
326 close(e);
327 }
328 }
329
330
331
332
333
334
335
336
337 public Promise<Message> callMethod(final Descriptors.MethodDescriptor method,
338 final PayloadCarryingRpcController controller, final Message request,
339 final Message responsePrototype, MetricsConnection.CallStats callStats) {
340 final AsyncCall call = new AsyncCall(channel.eventLoop(), client.callIdCnt.getAndIncrement(),
341 method, request, controller, responsePrototype, callStats);
342 controller.notifyOnCancel(new RpcCallback<Object>() {
343 @Override
344 public void run(Object parameter) {
345
346 synchronized (pendingCalls) {
347 pendingCalls.remove(call.id);
348 }
349 }
350 });
351
352 if (controller.isCanceled()) {
353
354 call.cancel(true);
355 return call;
356 }
357
358 synchronized (pendingCalls) {
359 if (closed) {
360 Promise<Message> promise = channel.eventLoop().newPromise();
361 promise.setFailure(new ConnectException());
362 return promise;
363 }
364 pendingCalls.put(call.id, call);
365
366 if (cleanupTimer == null && call.getRpcTimeout() > 0) {
367 cleanupTimer = client.newTimeout(timeoutTask, call.getRpcTimeout(), TimeUnit.MILLISECONDS);
368 }
369 if (!connected) {
370 return call;
371 }
372 }
373 writeRequest(call);
374 return call;
375 }
376
377 AsyncCall removePendingCall(int id) {
378 synchronized (pendingCalls) {
379 return pendingCalls.remove(id);
380 }
381 }
382
383
384
385
386
387
388
389 private ChannelFuture writeChannelHeader(Channel channel) throws IOException {
390 RPCProtos.ConnectionHeader header = getChannelHeader(authMethod);
391 int totalSize = IPCUtil.getTotalSizeWhenWrittenDelimited(header);
392
393 ByteBuf b = channel.alloc().directBuffer(totalSize);
394
395 b.writeInt(header.getSerializedSize());
396 b.writeBytes(header.toByteArray());
397
398 return channel.writeAndFlush(b);
399 }
400
401 private byte[] getChannelHeaderBytes(AuthMethod authMethod) {
402 RPCProtos.ConnectionHeader header = getChannelHeader(authMethod);
403 ByteBuffer b = ByteBuffer.allocate(header.getSerializedSize() + 4);
404 b.putInt(header.getSerializedSize());
405 b.put(header.toByteArray());
406 return b.array();
407 }
408
409 private RPCProtos.ConnectionHeader getChannelHeader(AuthMethod authMethod) {
410 RPCProtos.ConnectionHeader.Builder headerBuilder = RPCProtos.ConnectionHeader.newBuilder()
411 .setServiceName(serviceName);
412
413 RPCProtos.UserInformation userInfoPB = buildUserInfo(ticket.getUGI(), authMethod);
414 if (userInfoPB != null) {
415 headerBuilder.setUserInfo(userInfoPB);
416 }
417
418 if (client.codec != null) {
419 headerBuilder.setCellBlockCodecClass(client.codec.getClass().getCanonicalName());
420 }
421 if (client.compressor != null) {
422 headerBuilder.setCellBlockCompressorClass(client.compressor.getClass().getCanonicalName());
423 }
424
425 headerBuilder.setVersionInfo(ProtobufUtil.getVersionInfo());
426 return headerBuilder.build();
427 }
428
429
430
431
432
433 private void writeRequest(final AsyncCall call) {
434 try {
435 final RPCProtos.RequestHeader.Builder requestHeaderBuilder = RPCProtos.RequestHeader
436 .newBuilder();
437 requestHeaderBuilder.setCallId(call.id).setMethodName(call.method.getName())
438 .setRequestParam(call.param != null);
439
440 if (Trace.isTracing()) {
441 Span s = Trace.currentSpan();
442 requestHeaderBuilder.setTraceInfo(TracingProtos.RPCTInfo.newBuilder()
443 .setParentId(s.getSpanId()).setTraceId(s.getTraceId()));
444 }
445
446 ByteBuffer cellBlock = client.buildCellBlock(call.controller.cellScanner());
447 if (cellBlock != null) {
448 final RPCProtos.CellBlockMeta.Builder cellBlockBuilder = RPCProtos.CellBlockMeta
449 .newBuilder();
450 cellBlockBuilder.setLength(cellBlock.limit());
451 requestHeaderBuilder.setCellBlockMeta(cellBlockBuilder.build());
452 }
453
454 if (call.controller.getPriority() != PayloadCarryingRpcController.PRIORITY_UNSET) {
455 requestHeaderBuilder.setPriority(call.controller.getPriority());
456 }
457 requestHeaderBuilder.setTimeout(call.rpcTimeout > Integer.MAX_VALUE ?
458 Integer.MAX_VALUE : (int)call.rpcTimeout);
459 RPCProtos.RequestHeader rh = requestHeaderBuilder.build();
460
461 int totalSize = IPCUtil.getTotalSizeWhenWrittenDelimited(rh, call.param);
462 if (cellBlock != null) {
463 totalSize += cellBlock.remaining();
464 }
465
466 ByteBuf b = channel.alloc().directBuffer(4 + totalSize);
467 try (ByteBufOutputStream out = new ByteBufOutputStream(b)) {
468 call.callStats.setRequestSizeBytes(IPCUtil.write(out, rh, call.param, cellBlock));
469 }
470
471 channel.writeAndFlush(b).addListener(new CallWriteListener(this, call.id));
472 } catch (IOException e) {
473 close(e);
474 }
475 }
476
477
478
479
480
481 private void setupAuthorization() throws IOException {
482 SecurityInfo securityInfo = SecurityInfo.getInfo(serviceName);
483 this.useSasl = client.userProvider.isHBaseSecurityEnabled();
484
485 this.token = null;
486 if (useSasl && securityInfo != null) {
487 AuthenticationProtos.TokenIdentifier.Kind tokenKind = securityInfo.getTokenKind();
488 if (tokenKind != null) {
489 TokenSelector<? extends TokenIdentifier> tokenSelector = TOKEN_HANDDLERS.get(tokenKind);
490 if (tokenSelector != null) {
491 token = tokenSelector.selectToken(new Text(client.clusterId),
492 ticket.getUGI().getTokens());
493 } else if (LOG.isDebugEnabled()) {
494 LOG.debug("No token selector found for type " + tokenKind);
495 }
496 }
497 String serverKey = securityInfo.getServerPrincipal();
498 if (serverKey == null) {
499 throw new IOException("Can't obtain server Kerberos config key from SecurityInfo");
500 }
501 this.serverPrincipal = SecurityUtil.getServerPrincipal(client.conf.get(serverKey),
502 address.getAddress().getCanonicalHostName().toLowerCase());
503 if (LOG.isDebugEnabled()) {
504 LOG.debug("RPC Server Kerberos principal name for service=" + serviceName + " is "
505 + serverPrincipal);
506 }
507 }
508
509 if (!useSasl) {
510 authMethod = AuthMethod.SIMPLE;
511 } else if (token != null) {
512 authMethod = AuthMethod.DIGEST;
513 } else {
514 authMethod = AuthMethod.KERBEROS;
515 }
516
517 if (LOG.isDebugEnabled()) {
518 LOG.debug(
519 "Use " + authMethod + " authentication for service " + serviceName + ", sasl=" + useSasl);
520 }
521 reloginMaxBackoff = client.conf.getInt("hbase.security.relogin.maxbackoff", 5000);
522 }
523
524
525
526
527
528
529
530 private RPCProtos.UserInformation buildUserInfo(UserGroupInformation ugi, AuthMethod authMethod) {
531 if (ugi == null || authMethod == AuthMethod.DIGEST) {
532
533 return null;
534 }
535 RPCProtos.UserInformation.Builder userInfoPB = RPCProtos.UserInformation.newBuilder();
536 if (authMethod == AuthMethod.KERBEROS) {
537
538 userInfoPB.setEffectiveUser(ugi.getUserName());
539 } else if (authMethod == AuthMethod.SIMPLE) {
540
541 userInfoPB.setEffectiveUser(ugi.getUserName());
542 if (ugi.getRealUser() != null) {
543 userInfoPB.setRealUser(ugi.getRealUser().getUserName());
544 }
545 }
546 return userInfoPB.build();
547 }
548
549
550
551
552
553
554 private void createPreamble(ByteBuf byteBuf, AuthMethod authMethod) {
555 byteBuf.writeBytes(HConstants.RPC_HEADER);
556 byteBuf.writeByte(HConstants.RPC_CURRENT_VERSION);
557 byteBuf.writeByte(authMethod.code);
558 }
559
560 private void close0(Throwable e) {
561 List<AsyncCall> toCleanup;
562 synchronized (pendingCalls) {
563 if (closed) {
564 return;
565 }
566 closed = true;
567 toCleanup = new ArrayList<AsyncCall>(pendingCalls.values());
568 pendingCalls.clear();
569 }
570 IOException closeException = null;
571 if (e != null) {
572 if (e instanceof IOException) {
573 closeException = (IOException) e;
574 } else {
575 closeException = new IOException(e);
576 }
577 }
578
579 if (LOG.isDebugEnabled() && closeException != null) {
580 LOG.debug(name + ": closing ipc connection to " + address, closeException);
581 }
582 if (cleanupTimer != null) {
583 cleanupTimer.cancel();
584 cleanupTimer = null;
585 }
586 for (AsyncCall call : toCleanup) {
587 call.setFailed(closeException != null ? closeException
588 : new ConnectionClosingException(
589 "Call id=" + call.id + " on server " + address + " aborted: connection is closing"));
590 }
591 channel.disconnect().addListener(ChannelFutureListener.CLOSE);
592 if (LOG.isDebugEnabled()) {
593 LOG.debug(name + ": closed");
594 }
595 }
596
597
598
599
600
601 public void close(final Throwable e) {
602 client.removeConnection(this);
603
604
605 if (channel.eventLoop().inEventLoop()) {
606 close0(e);
607 } else {
608 channel.eventLoop().execute(new Runnable() {
609 @Override
610 public void run() {
611 close0(e);
612 }
613 });
614 }
615 }
616
617
618
619
620 private void cleanupCalls() {
621 List<AsyncCall> toCleanup = new ArrayList<AsyncCall>();
622 long currentTime = EnvironmentEdgeManager.currentTime();
623 long nextCleanupTaskDelay = -1L;
624 synchronized (pendingCalls) {
625 for (Iterator<AsyncCall> iter = pendingCalls.values().iterator(); iter.hasNext();) {
626 AsyncCall call = iter.next();
627 long timeout = call.getRpcTimeout();
628 if (timeout > 0) {
629 if (currentTime - call.getStartTime() >= timeout) {
630 iter.remove();
631 toCleanup.add(call);
632 } else {
633 if (nextCleanupTaskDelay < 0 || timeout < nextCleanupTaskDelay) {
634 nextCleanupTaskDelay = timeout;
635 }
636 }
637 }
638 }
639 if (nextCleanupTaskDelay > 0) {
640 cleanupTimer = client.newTimeout(timeoutTask, nextCleanupTaskDelay, TimeUnit.MILLISECONDS);
641 } else {
642 cleanupTimer = null;
643 }
644 }
645 for (AsyncCall call : toCleanup) {
646 call.setFailed(new CallTimeoutException("Call id=" + call.id + ", waitTime="
647 + (currentTime - call.getStartTime()) + ", rpcTimeout=" + call.getRpcTimeout()));
648 }
649 }
650
651
652
653
654
655 public boolean isAlive() {
656 return channel.isOpen();
657 }
658
659
660
661
662
663
664 private synchronized boolean shouldAuthenticateOverKrb() throws IOException {
665 UserGroupInformation loginUser = UserGroupInformation.getLoginUser();
666 UserGroupInformation currentUser = UserGroupInformation.getCurrentUser();
667 UserGroupInformation realUser = currentUser.getRealUser();
668 return authMethod == AuthMethod.KERBEROS && loginUser != null &&
669
670 loginUser.hasKerberosCredentials() &&
671
672
673 (loginUser.equals(currentUser) || loginUser.equals(realUser));
674 }
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695 private void handleSaslConnectionFailure(final int currRetries, final Throwable ex,
696 final UserGroupInformation user) throws IOException, InterruptedException {
697 user.doAs(new PrivilegedExceptionAction<Void>() {
698 @Override
699 public Void run() throws IOException, InterruptedException {
700 if (shouldAuthenticateOverKrb()) {
701 if (currRetries < MAX_SASL_RETRIES) {
702 LOG.debug("Exception encountered while connecting to the server : " + ex);
703
704 if (UserGroupInformation.isLoginKeytabBased()) {
705 UserGroupInformation.getLoginUser().reloginFromKeytab();
706 } else {
707 UserGroupInformation.getLoginUser().reloginFromTicketCache();
708 }
709
710
711 return null;
712 } else {
713 String msg = "Couldn't setup connection for "
714 + UserGroupInformation.getLoginUser().getUserName() + " to " + serverPrincipal;
715 LOG.warn(msg);
716 throw (IOException) new IOException(msg).initCause(ex);
717 }
718 } else {
719 LOG.warn("Exception encountered while connecting to " + "the server : " + ex);
720 }
721 if (ex instanceof RemoteException) {
722 throw (RemoteException) ex;
723 }
724 if (ex instanceof SaslException) {
725 String msg = "SASL authentication failed."
726 + " The most likely cause is missing or invalid credentials." + " Consider 'kinit'.";
727 LOG.fatal(msg, ex);
728 throw new RuntimeException(msg, ex);
729 }
730 throw new IOException(ex);
731 }
732 });
733 }
734
735 public int getConnectionHashCode() {
736 return ConnectionId.hashCode(ticket, serviceName, address);
737 }
738
739 @Override
740 public int hashCode() {
741 return getConnectionHashCode();
742 }
743
744 @Override
745 public boolean equals(Object obj) {
746 if (obj instanceof AsyncRpcChannel) {
747 AsyncRpcChannel channel = (AsyncRpcChannel) obj;
748 return channel.hashCode() == obj.hashCode();
749 }
750 return false;
751 }
752
753 @Override
754 public String toString() {
755 return this.address.toString() + "/" + this.serviceName + "/" + this.ticket;
756 }
757
758
759
760
761 private static final class CallWriteListener implements ChannelFutureListener {
762 private final AsyncRpcChannel rpcChannel;
763 private final int id;
764
765 public CallWriteListener(AsyncRpcChannel asyncRpcChannel, int id) {
766 this.rpcChannel = asyncRpcChannel;
767 this.id = id;
768 }
769
770 @Override
771 public void operationComplete(ChannelFuture future) throws Exception {
772 if (!future.isSuccess()) {
773 AsyncCall call = rpcChannel.removePendingCall(id);
774 if (call != null) {
775 if (future.cause() instanceof IOException) {
776 call.setFailed((IOException) future.cause());
777 } else {
778 call.setFailed(new IOException(future.cause()));
779 }
780 }
781 }
782 }
783 }
784 }