@Override
protected void doStart() {

    try {

        log.load();

        RaftServiceEndpoint endpoint = new RaftServiceEndpoint(ctx);
        Service replicaService = RaftProto.RaftService.newReflectiveService(endpoint);
        rpcServer.registerService(replicaService);
        rpcServer.startAsync().addListener(new Listener() {
            @Override
            public void running() {
                ctx.setState(RaftStateContext.StateType.FOLLOWER);
            }
        }, MoreExecutors.sameThreadExecutor());
        rpcServer.awaitRunning();

        notifyStarted();

    } catch (Exception e) {
        notifyFailed(e);
    }

}
 

/**
 * Handle the blocking RPC request by forwarding it to the correct
 * service/method.
 *
 * @throws RpcException If there was some error executing the RPC.
 */
public SocketRpcProtos.Response doBlockingRpc(
    SocketRpcProtos.Request rpcRequest) throws RpcException {
  // Get the service, first try BlockingService
  BlockingService blockingService = blockingServiceMap.get(
      rpcRequest.getServiceName());
  if (blockingService != null) {
    return forwardToBlockingService(rpcRequest, blockingService);
  }

  // Now try Service
  Service service = serviceMap.get(rpcRequest.getServiceName());
  if (service == null) {
    throw new RpcException(ErrorReason.SERVICE_NOT_FOUND,
        "Could not find service: " + rpcRequest.getServiceName(), null);
  }

  // Call service using an instant callback
  Callback<Message> callback = new Callback<Message>();
  SocketRpcController socketController = new SocketRpcController();
  forwardToService(rpcRequest, callback, service, socketController);

  // Build and return response (callback invocation is optional)
  return createRpcResponse(callback.response, callback.invoked,
      socketController);
}
 

private void forwardToService(SocketRpcProtos.Request rpcRequest,
    RpcCallback<Message> callback, Service service,
    RpcController socketController) throws RpcException {
  // Get matching method
  MethodDescriptor method = getMethod(rpcRequest,
      service.getDescriptorForType());

  // Create request for method
  Message request = getRequestProto(rpcRequest,
      service.getRequestPrototype(method));

  // Call method
  try {
    service.callMethod(method, socketController, request, callback);
  } catch (RuntimeException e) {
    throw new RpcException(ErrorReason.RPC_ERROR,
        "Error running method " + method.getFullName(), e);
  }
}
 

public AsyncRpcServer(final Class<?> protocol,
                      final Object instance,
                      final InetSocketAddress bindAddress,
                      final int threads)
    throws Exception {
  super(protocol.getSimpleName(), bindAddress);

  String serviceClassName = protocol.getName() + "$" +
      protocol.getSimpleName() + "Service";
  Class<?> serviceClass = Class.forName(serviceClassName);
  Class<?> interfaceClass = Class.forName(serviceClassName + "$Interface");
  Method method = serviceClass.getMethod("newReflectiveService", interfaceClass);
  this.service = (Service) method.invoke(null, instance);

  this.initializer = new ProtoServerChannelInitializer(new ServerHandler(), RpcRequest.getDefaultInstance());
  super.init(this.initializer, threads);
}
 

/**
	 * Register a remote RPC service to system.
	 * @param service
	 */
	public void registerRpcService(Service service) {
		String serviceName = service.getDescriptorForType().getFullName();
		rpcServiceMap.put(serviceName, service);
//		List<MethodDescriptor> methods = service.getDescriptorForType().getMethods();
		if ( logger.isInfoEnabled() ) {
			logger.info("register remote service: {}", serviceName);
		}
//		for ( MethodDescriptor method : methods ) {
//			String methodName = method.getName();
//			String remoteServiceName = StringUtil.concat(serviceName, DOT, methodName);
//			rpcServiceMap.put(remoteServiceName, service);
//			if ( log.isInfoEnabled() ) {
//				log.info("register remote service: " + remoteServiceName );
//			}
//		}
	}
 

public AsyncRpcServer(final Class<?> protocol,
                      final Object instance,
                      final InetSocketAddress bindAddress,
                      final int workerNum)
    throws Exception {
  super(protocol.getSimpleName(), bindAddress);

  String serviceClassName = protocol.getName() + "$" +
      protocol.getSimpleName() + "Service";
  Class<?> serviceClass = Class.forName(serviceClassName);
  Class<?> interfaceClass = Class.forName(serviceClassName + "$Interface");
  Method method = serviceClass.getMethod("newReflectiveService", interfaceClass);
  this.service = (Service) method.invoke(null, instance);

  ServerHandler handler = new ServerHandler();
  this.pipeline = new ProtoPipelineFactory(handler,
      RpcRequest.getDefaultInstance());
  super.init(this.pipeline, workerNum);
}
 

/**
 * {@inheritDoc}
 */
@Override
public <T extends Service, R> Map<byte[], R> coprocessorService(final Class<T> service,
                                                                byte[] startKey, byte[] endKey, final Batch.Call<T, R> callable)
        throws ServiceException {
    throw new RuntimeException(this.getClass() + " does NOT implement this method.");
}
 

/**
 * {@inheritDoc}
 */
@Override
public <T extends Service, R> void coprocessorService(final Class<T> service,
                                                      byte[] startKey, byte[] endKey, final Batch.Call<T, R> callable,
                                                      final Batch.Callback<R> callback) throws ServiceException {
    throw new RuntimeException(this.getClass() + " does NOT implement this method.");
}
 

/**
 * Register all methods provided by the given service type.
 */
public void registerService(Service service) {
  synchronized (registeredServices) {
    for (MethodDescriptor methodDescriptor : service.getDescriptorForType().getMethods()) {
      registeredServices.put(methodDescriptor.getInputType(),
          new RegisteredServiceMethod(service, methodDescriptor));
    }
  }
}
 

@Override
public Message preEndpointInvocation(ObserverContext<RegionCoprocessorEnvironment> ctx,
									 Service service, String methodName, Message request) throws IOException {
	// Don't intercept calls to our own AccessControlService, we check for
	// appropriate permissions in the service handlers
	if (shouldCheckExecPermission && !(service instanceof AccessControlService)) {
		requirePermission(ctx,
				"invoke(" + service.getDescriptorForType().getName() + "." + methodName + ")",
				getTableName(ctx.getEnvironment()), null, null,
				Action.EXEC);
	}
	return request;
}
 

@Override
public <T extends Service, R> void coprocessorService(Class<T> service, byte[] startKey,
        byte[] endKey, Call<T, R> callable, Callback<R> callback) throws ServiceException,
        Throwable {
    delegate.coprocessorService(service, startKey, endKey, callable, callback);
    
}
 

/**
 * Register all methods provided by the given service type.
 */
public void registerService(Service service) {
  synchronized (registeredServices) {
    for (MethodDescriptor methodDescriptor : service.getDescriptorForType().getMethods()) {
      registeredServices.put(methodDescriptor.getInputType(),
          new RegisteredServiceMethod(service, methodDescriptor));
    }
  }
}
 

public <T extends Service,V> Map<byte[],V> coprocessorExec(
        Class<T> serviceClass,
        byte[] startKey,
        byte[] endKey,
        final Batch.Call<T,V> callable) throws Throwable {
    return delegate.coprocessorExec(serviceClass, startKey, endKey, callable);
}
 

public <T extends Service,V> Map<byte[],V> coprocessorExec(
        Class<T> serviceClass,
        byte[] startKey,
        byte[] endKey,
        Batch.Call<T,V> call) throws Throwable{
    return table.coprocessorService(serviceClass, startKey, endKey, call);
}
 

@Override
public Service getService() {
    return this;
}
 

@Override
public <T extends Service, R> Map<byte[], R> coprocessorService(Class<T> service, byte[] startKey, byte[] endKey,
        Batch.Call<T, R> callable) throws ServiceException, Throwable {
    throw new NotImplementedException();

}
 

@Override
public <T extends Service, R> void coprocessorService(Class<T> service, byte[] startKey, byte[] endKey,
        Batch.Call<T, R> callable, Batch.Callback<R> callback) throws ServiceException, Throwable {
    throw new NotImplementedException();

}
 

private void configureRpcServer() {
  RaftServiceEndpoint endpoint = new RaftServiceEndpoint(ctx);
  Service replicaService = RaftProto.RaftService.newReflectiveService(endpoint);
  rpcServer.registerService(replicaService);
}
 

public CupidRpcChannelProxyTest(Service service) {
    this.service = service;
}
 

public void registerService(Service service) {
    handler.registerService(service);
}
 

public void unregisterService(Service service) {
    handler.unregisterService(service);
}
 

@Override
public <T extends Service, R> Map<byte[], R> coprocessorService(Class<T> service, byte[] startKey, byte[] endKey,
                                                                Batch.Call<T, R> callable)
  throws ServiceException, Throwable {
  return hTable.coprocessorService(service, startKey, endKey, callable);
}
 

@Override
public <T extends Service, R> void coprocessorService(Class<T> service, byte[] startKey, byte[] endKey,
                                                      Batch.Call<T, R> callable, Batch.Callback<R> callback)
  throws ServiceException, Throwable {
  hTable.coprocessorService(service, startKey, endKey, callable, callback);
}
 

@Override
public <T extends Service, R> Map<byte[], R> coprocessorService(Class<T> service, byte[] startKey, byte[] endKey,
                                                                Batch.Call<T, R> callable)
  throws ServiceException, Throwable {
  return hTable.coprocessorService(service, startKey, endKey, callable);
}
 

@Override
public <T extends Service, R> void coprocessorService(Class<T> service, byte[] startKey, byte[] endKey,
                                                      Batch.Call<T, R> callable, Batch.Callback<R> callback)
  throws ServiceException, Throwable {
  hTable.coprocessorService(service, startKey, endKey, callable, callback);
}
 

@Override
public <T extends Service, R> Map<byte[], R> coprocessorService(Class<T> service, byte[] startKey, byte[] endKey,
                                                                Batch.Call<T, R> callable)
  throws ServiceException, Throwable {
  return hTable.coprocessorService(service, startKey, endKey, callable);
}
 

@Override
public <T extends Service, R> void coprocessorService(Class<T> service, byte[] startKey, byte[] endKey,
                                                      Batch.Call<T, R> callable, Batch.Callback<R> callback)
  throws ServiceException, Throwable {
  hTable.coprocessorService(service, startKey, endKey, callable, callback);
}
 

@Override
public <T extends Service, R> Map<byte[], R> coprocessorService(Class<T> service, byte[] startKey,
    byte[] endKey, Batch.Call<T, R> callable) throws ServiceException, Throwable {
  return hTable.coprocessorService(service, startKey, endKey, callable);
}
 

@Override
public <T extends Service, R> void coprocessorService(Class<T> service, byte[] startKey,
    byte[] endKey, Batch.Call<T, R> callable, Batch.Callback<R> callback)
    throws ServiceException, Throwable {
  hTable.coprocessorService(service, startKey, endKey, callable, callback);
}
 

public void setService(Service service) {
    this.service = service;
}