int main(int argc,char* argv[]) {
   EchoServiceImpl echo_service;
   RpcManager rpcman;
   ...

我找不到如何访问RpcManager及其位置.我对protobuf和protoserv dir做了一个grep无济于事.在.java包中只有RpcManager.也许它只是一个非supoorted版本,即使它在C中有例子.

任何帮助将不胜感激.即使是稳定/简单的C protobuf实现的链接也会很棒.

有关如何构建处理原始请求和响应的HTTP服务器,请检查pbrpc.

有关如何处理请求和生成响应(构建RPC系统),请查看libpbrpc.

每个项目只包含几个源文件,它们不会花费很长时间来阅读.

/**
 * Grant permissions globally to the given user. Will wait until all active
 * AccessController instances have updated their permissions caches or will
 * throw an exception upon timeout (10 seconds).
 */
public static void grantGlobal(final hbasetestingutility util,final String user,final Permission.Action... actions) throws Exception {
  SecureTestUtil.updateACLs(util,new Callable<Void>() {
    @Override
    public Void call() throws Exception {
      try (Connection connection = ConnectionFactory.createConnection(util.getConfiguration())) {
        try (Table acl = connection.getTable(AccessControlLists.ACL_TABLE_NAME)) {
          BlockingRpcChannel service = acl.coprocessorService(HConstants.EMPTY_START_ROW);
          AccessControlService.BlockingInterface protocol =
              AccessControlService.newBlockingStub(service);
          ProtobufUtil.grant(null,protocol,user,actions);
        }
      }
      return null;
    }
  });
}

/**
 * Revoke permissions globally from the given user. Will wait until all active
 * AccessController instances have updated their permissions caches or will
 * throw an exception upon timeout (10 seconds).
 */
public static void revokeGlobal(final hbasetestingutility util,new Callable<Void>() {
    @Override
    public Void call() throws Exception {
      try (Connection connection = ConnectionFactory.createConnection(util.getConfiguration())) {
        try (Table acl = connection.getTable(AccessControlLists.ACL_TABLE_NAME)) {
          BlockingRpcChannel service = acl.coprocessorService(HConstants.EMPTY_START_ROW);
          AccessControlService.BlockingInterface protocol =
              AccessControlService.newBlockingStub(service);
          ProtobufUtil.revoke(null,actions);
        }
      }
      return null;
    }
  });
}

/**
 * Grant permissions on a namespace to the given user. Will wait until all active
 * AccessController instances have updated their permissions caches or will
 * throw an exception upon timeout (10 seconds).
 */
public static void grantOnNamespace(final hbasetestingutility util,final String namespace,namespace,actions);
        }
      }
      return null;
    }
  });
}

/**
 * Revoke permissions on a namespace from the given user. Will wait until all active
 * AccessController instances have updated their permissions caches or will
 * throw an exception upon timeout (10 seconds).
 */
public static void revokeFromNamespace(final hbasetestingutility util,actions);
        }
      }
      return null;
    }
  });
}

/**
 * Grant permissions on a table to the given user. Will wait until all active
 * AccessController instances have updated their permissions caches or will
 * throw an exception upon timeout (10 seconds).
 */
public static void grantOnTable(final hbasetestingutility util,final TableName table,final byte[] family,final byte[] qualifier,table,family,qualifier,actions);
        }
      }
      return null;
    }
  });
}

/**
 * Revoke permissions on a table from the given user. Will wait until all active
 * AccessController instances have updated their permissions caches or will
 * throw an exception upon timeout (10 seconds).
 */
public static void revokeFromTable(final hbasetestingutility util,actions);
        }
      }
      return null;
    }
  });
}

public static void checkGlobalPerms(hbasetestingutility testUtil,Permission.Action... actions)
    throws IOException {
  Permission[] perms = new Permission[actions.length];
  for (int i = 0; i < actions.length; i++) {
    perms[i] = new Permission(actions[i]);
  }
  CheckPermissionsRequest.Builder request = CheckPermissionsRequest.newBuilder();
  for (Action a : actions) {
    request.addPermission(AccessControlProtos.Permission.newBuilder()
        .setType(AccessControlProtos.Permission.Type.Global)
        .setGlobalPermission(
            AccessControlProtos.GlobalPermission.newBuilder()
                .addAction(ProtobufUtil.toPermissionAction(a)).build()));
  }
  try(Connection conn = ConnectionFactory.createConnection(testUtil.getConfiguration());
      Table acl = conn.getTable(AccessControlLists.ACL_TABLE_NAME)) {
    BlockingRpcChannel channel = acl.coprocessorService(new byte[0]);
    AccessControlService.BlockingInterface protocol =
      AccessControlService.newBlockingStub(channel);
    try {
      protocol.checkPermissions(null,request.build());
    } catch (ServiceException se) {
      ProtobufUtil.toIOException(se);
    }
  }
}

@Test (timeout=180000)
public void testGlobalPermissionList() throws Exception {
  List<UserPermission> perms;
  Table acl = systemUserConnection.getTable(AccessControlLists.ACL_TABLE_NAME);
  try {
    BlockingRpcChannel service = acl.coprocessorService(HConstants.EMPTY_START_ROW);
    AccessControlService.BlockingInterface protocol =
      AccessControlService.newBlockingStub(service);
    perms = ProtobufUtil.getUserPermissions(null,protocol);
  } finally {
    acl.close();
  }
  UserPermission adminPerm = new UserPermission(Bytes.toBytes(USER_ADMIN.getShortName()),AccessControlLists.ACL_TABLE_NAME,null,Bytes.toBytes("ACRW"));
  assertTrue("Only global users and user admin has permission on table _acl_ per setup",perms.size() == 5 && hasFoundUserPermission(adminPerm,perms));
}

@Override
// nothing is done w/ the 'master' parameter.  It is ignored.
public AdminService.BlockingInterface getAdmin(final ServerName serverName,final boolean master)
throws IOException {
  if (isDeadServer(serverName)) {
    throw new RegionServerStoppedException(serverName + " is dead.");
  }
  String key = getStubKey(AdminService.BlockingInterface.class.getName(),serverName.getHostname(),serverName.getPort(),this.hostnamesCanChange);
  this.connectionLock.putIfAbsent(key,key);
  AdminService.BlockingInterface stub = null;
  synchronized (this.connectionLock.get(key)) {
    stub = (AdminService.BlockingInterface)this.stubs.get(key);
    if (stub == null) {
      BlockingRpcChannel channel =
          this.rpcclient.createBlockingRpcChannel(serverName,rpcTimeout);
      stub = AdminService.newBlockingStub(channel);
      this.stubs.put(key,stub);
    }
  }
  return stub;
}

@Override
public ClientService.BlockingInterface getClient(final ServerName sn)
throws IOException {
  if (isDeadServer(sn)) {
    throw new RegionServerStoppedException(sn + " is dead.");
  }
  String key = getStubKey(ClientService.BlockingInterface.class.getName(),sn.getHostname(),sn.getPort(),key);
  ClientService.BlockingInterface stub = null;
  synchronized (this.connectionLock.get(key)) {
    stub = (ClientService.BlockingInterface)this.stubs.get(key);
    if (stub == null) {
      BlockingRpcChannel channel =
          this.rpcclient.createBlockingRpcChannel(sn,rpcTimeout);
      stub = ClientService.newBlockingStub(channel);
      // In old days,after getting stub/proxy,we'd make a call.  We are not doing that here.
      // Just fail on first actual call rather than in here on setup.
      this.stubs.put(key,stub);
    }
  }
  return stub;
}

/**
 * Grant permissions globally to the given user. Will wait until all active
 * AccessController instances have updated their permissions caches or will
 * throw an exception upon timeout (10 seconds).
 */
public static void grantGlobal(final hbasetestingutility util,new Callable<Void>() {
    @Override
    public Void call() throws Exception {
      try (Connection connection = ConnectionFactory.createConnection(util.getConfiguration())) {
        try (Table acl = connection.getTable(AccessControlLists.ACL_TABLE_NAME)) {
          BlockingRpcChannel service = acl.coprocessorService(HConstants.EMPTY_START_ROW);
          AccessControlService.BlockingInterface protocol =
              AccessControlService.newBlockingStub(service);
          ProtobufUtil.grant(protocol,actions);
        }
      }
      return null;
    }
  });
}

/**
 * Revoke permissions globally from the given user. Will wait until all active
 * AccessController instances have updated their permissions caches or will
 * throw an exception upon timeout (10 seconds).
 */
public static void revokeGlobal(final hbasetestingutility util,new Callable<Void>() {
    @Override
    public Void call() throws Exception {
      try (Connection connection = ConnectionFactory.createConnection(util.getConfiguration())) {
        try (Table acl = connection.getTable(AccessControlLists.ACL_TABLE_NAME)) {
          BlockingRpcChannel service = acl.coprocessorService(HConstants.EMPTY_START_ROW);
          AccessControlService.BlockingInterface protocol =
              AccessControlService.newBlockingStub(service);
          ProtobufUtil.revoke(protocol,actions);
        }
      }
      return null;
    }
  });
}

/**
 * Grant permissions on a namespace to the given user. Will wait until all active
 * AccessController instances have updated their permissions caches or will
 * throw an exception upon timeout (10 seconds).
 */
public static void grantOnNamespace(final hbasetestingutility util,actions);
        }
      }
      return null;
    }
  });
}

/**
 * Revoke permissions on a namespace from the given user. Will wait until all active
 * AccessController instances have updated their permissions caches or will
 * throw an exception upon timeout (10 seconds).
 */
public static void revokeFromNamespace(final hbasetestingutility util,actions);
        }
      }
      return null;
    }
  });
}

/**
 * Grant permissions on a table to the given user. Will wait until all active
 * AccessController instances have updated their permissions caches or will
 * throw an exception upon timeout (10 seconds).
 */
public static void grantOnTable(final hbasetestingutility util,actions);
        }
      }
      return null;
    }
  });
}

/**
 * Revoke permissions on a table from the given user. Will wait until all active
 * AccessController instances have updated their permissions caches or will
 * throw an exception upon timeout (10 seconds).
 */
public static void revokeFromTable(final hbasetestingutility util,actions);
        }
      }
      return null;
    }
  });
}

@Test
public void testGlobalPermissionList() throws Exception {
  List<UserPermission> perms;
  Table acl = new HTable(conf,AccessControlLists.ACL_TABLE_NAME);
  try {
    BlockingRpcChannel service = acl.coprocessorService(HConstants.EMPTY_START_ROW);
    AccessControlService.BlockingInterface protocol =
      AccessControlService.newBlockingStub(service);
    perms = ProtobufUtil.getUserPermissions(protocol);
  } finally {
    acl.close();
  }
  UserPermission adminPerm = new UserPermission(Bytes.toBytes(USER_ADMIN.getShortName()),Bytes.toBytes("ACRW"));
  assertTrue("Only user admin has permission on table _acl_ per setup",perms.size() == 1 && hasFoundUserPermission(adminPerm,perms));
}

@Override
// nothing is done w/ the 'master' parameter.  It is ignored.
public AdminService.BlockingInterface getAdmin(final ServerName serverName,final boolean master)
        throws IOException {
    if (isDeadServer(serverName)) {
        throw new RegionServerStoppedException(serverName + " is dead.");
    }
    String key = getStubKey(AdminService.BlockingInterface.class.getName(),serverName.getHostAndPort());
    this.connectionLock.putIfAbsent(key,key);
    AdminService.BlockingInterface stub = null;
    synchronized (this.connectionLock.get(key)) {
        stub = (AdminService.BlockingInterface) this.stubs.get(key);
        if (stub == null) {
            BlockingRpcChannel channel =
                    this.rpcclient.createBlockingRpcChannel(serverName,rpcTimeout);
            stub = AdminService.newBlockingStub(channel);
            this.stubs.put(key,stub);
        }
    }
    return stub;
}

@Override
public ClientService.BlockingInterface getClient(final ServerName sn)
        throws IOException {
    if (isDeadServer(sn)) {
        throw new RegionServerStoppedException(sn + " is dead.");
    }
    String key = getStubKey(ClientService.BlockingInterface.class.getName(),sn.getHostAndPort());
    this.connectionLock.putIfAbsent(key,key);
    ClientService.BlockingInterface stub = null;
    synchronized (this.connectionLock.get(key)) {
        stub = (ClientService.BlockingInterface) this.stubs.get(key);
        if (stub == null) {
            BlockingRpcChannel channel =
                    this.rpcclient.createBlockingRpcChannel(sn,rpcTimeout);
            stub = ClientService.newBlockingStub(channel);
            // In old days,we'd make a call.  We are not doing that here.
            // Just fail on first actual call rather than in here on setup.
            this.stubs.put(key,stub);
        }
    }
    return stub;
}

/**
 * Grant permissions globally to the given user. Will wait until all active
 * AccessController instances have updated their permissions caches or will
 * throw an exception upon timeout (10 seconds).
 */
public static void grantGlobal(final hbasetestingutility util,new Callable<Void>() {
    @Override
    public Void call() throws Exception {
      HTable acl = new HTable(util.getConfiguration(),AccessControlLists.ACL_TABLE_NAME);
      try {
        BlockingRpcChannel service = acl.coprocessorService(HConstants.EMPTY_START_ROW);
        AccessControlService.BlockingInterface protocol =
            AccessControlService.newBlockingStub(service);
        ProtobufUtil.grant(protocol,actions);
      } finally {
        acl.close();
      }
      return null;
    }
  });
}

/**
 * Revoke permissions globally from the given user. Will wait until all active
 * AccessController instances have updated their permissions caches or will
 * throw an exception upon timeout (10 seconds).
 */
public static void revokeGlobal(final hbasetestingutility util,AccessControlLists.ACL_TABLE_NAME);
      try {
        BlockingRpcChannel service = acl.coprocessorService(HConstants.EMPTY_START_ROW);
        AccessControlService.BlockingInterface protocol =
            AccessControlService.newBlockingStub(service);
        ProtobufUtil.revoke(protocol,actions);
      } finally {
        acl.close();
      }
      return null;
    }
  });
}

/**
 * Grant permissions on a namespace to the given user. Will wait until all active
 * AccessController instances have updated their permissions caches or will
 * throw an exception upon timeout (10 seconds).
 */
public static void grantOnNamespace(final hbasetestingutility util,actions);
      } finally {
        acl.close();
      }
      return null;
    }
  });
}

/**
 * Revoke permissions on a namespace from the given user. Will wait until all active
 * AccessController instances have updated their permissions caches or will
 * throw an exception upon timeout (10 seconds).
 */
public static void revokeFromNamespace(final hbasetestingutility util,actions);
      } finally {
        acl.close();
      }
      return null;
    }
  });
}

/**
 * Grant permissions on a table to the given user. Will wait until all active
 * AccessController instances have updated their permissions caches or will
 * throw an exception upon timeout (10 seconds).
 */
public static void grantOnTable(final hbasetestingutility util,actions);
      } finally {
        acl.close();
      }
      return null;
    }
  });
}

/**
 * Revoke permissions on a table from the given user. Will wait until all active
 * AccessController instances have updated their permissions caches or will
 * throw an exception upon timeout (10 seconds).
 */
public static void revokeFromTable(final hbasetestingutility util,actions);
      } finally {
        acl.close();
      }
      return null;
    }
  });
}

@Test
public void testGlobalPermissionList() throws Exception {
  List<UserPermission> perms;
  HTable acl = new HTable(conf,perms));
}

@Override
// nothing is done w/ the 'master' parameter.  It is ignored.
public AdminService.BlockingInterface getAdmin(final ServerName serverName,serverName.getHostAndPort());
  this.connectionLock.putIfAbsent(key,key);
  AdminService.BlockingInterface stub = null;
  synchronized (this.connectionLock.get(key)) {
    stub = (AdminService.BlockingInterface)this.stubs.get(key);
    if (stub == null) {
      BlockingRpcChannel channel = this.rpcclient.createBlockingRpcChannel(serverName,this.rpcTimeout);
      stub = AdminService.newBlockingStub(channel);
      this.stubs.put(key,stub);
    }
  }
  return stub;
}

@Override
public ClientService.BlockingInterface getClient(final ServerName sn)
throws IOException {
  if (isDeadServer(sn)) {
    throw new RegionServerStoppedException(sn + " is dead.");
  }
  String key = getStubKey(ClientService.BlockingInterface.class.getName(),sn.getHostAndPort());
  this.connectionLock.putIfAbsent(key,key);
  ClientService.BlockingInterface stub = null;
  synchronized (this.connectionLock.get(key)) {
    stub = (ClientService.BlockingInterface)this.stubs.get(key);
    if (stub == null) {
      BlockingRpcChannel channel = this.rpcclient.createBlockingRpcChannel(sn,this.rpcTimeout);
      stub = ClientService.newBlockingStub(channel);
      // In old days,stub);
    }
  }
  return stub;
}

/**
 * Grant permissions globally to the given user. Will wait until all active
 * AccessController instances have updated their permissions caches or will
 * throw an exception upon timeout (10 seconds).
 */
public static void grantGlobal(final hbasetestingutility util,new Callable<Void>() {
    @Override
    public Void call() throws Exception {
      try (Connection connection = ConnectionFactory.createConnection(util.getConfiguration())) {
        try (Table acl = connection.getTable(AccessControlLists.ACL_TABLE_NAME)) {
          BlockingRpcChannel service = acl.coprocessorService(HConstants.EMPTY_START_ROW);
          AccessControlService.BlockingInterface protocol =
              AccessControlService.newBlockingStub(service);
          AccessControlUtil.grant(null,false,actions);
        }
      }
      return null;
    }
  });
}

/**
 * Revoke permissions globally from the given user. Will wait until all active
 * AccessController instances have updated their permissions caches or will
 * throw an exception upon timeout (10 seconds).
 */
public static void revokeGlobal(final hbasetestingutility util,new Callable<Void>() {
    @Override
    public Void call() throws Exception {
      try (Connection connection = ConnectionFactory.createConnection(util.getConfiguration())) {
        try (Table acl = connection.getTable(AccessControlLists.ACL_TABLE_NAME)) {
          BlockingRpcChannel service = acl.coprocessorService(HConstants.EMPTY_START_ROW);
          AccessControlService.BlockingInterface protocol =
              AccessControlService.newBlockingStub(service);
          AccessControlUtil.revoke(null,actions);
        }
      }
      return null;
    }
  });
}

/**
 * Grant permissions on a namespace to the given user. Will wait until all active
 * AccessController instances have updated their permissions caches or will
 * throw an exception upon timeout (10 seconds).
 */
public static void grantOnNamespace(final hbasetestingutility util,actions);
        }
      }
      return null;
    }
  });
}

/**
 * Revoke permissions on a namespace from the given user. Will wait until all active
 * AccessController instances have updated their permissions caches or will
 * throw an exception upon timeout (10 seconds).
 */
public static void revokeFromNamespace(final hbasetestingutility util,actions);
        }
      }
      return null;
    }
  });
}

/**
 * Grant permissions on a table to the given user. Will wait until all active
 * AccessController instances have updated their permissions caches or will
 * throw an exception upon timeout (10 seconds).
 */
public static void grantOnTable(final hbasetestingutility util,actions);
        }
      }
      return null;
    }
  });
}

/**
 * Revoke permissions on a table from the given user. Will wait until all active
 * AccessController instances have updated their permissions caches or will
 * throw an exception upon timeout (10 seconds).
 */
public static void revokeFromTable(final hbasetestingutility util,actions);
        }
      }
      return null;
    }
  });
}

public static void checkGlobalPerms(hbasetestingutility testUtil,Permission.Action... actions)
    throws IOException {
  Permission[] perms = new Permission[actions.length];
  for (int i = 0; i < actions.length; i++) {
    perms[i] = new Permission(actions[i]);
  }
  CheckPermissionsRequest.Builder request = CheckPermissionsRequest.newBuilder();
  for (Action a : actions) {
    request.addPermission(AccessControlProtos.Permission.newBuilder()
        .setType(AccessControlProtos.Permission.Type.Global)
        .setGlobalPermission(
            AccessControlProtos.GlobalPermission.newBuilder()
                .addAction(AccessControlUtil.toPermissionAction(a)).build()));
  }
  try(Connection conn = ConnectionFactory.createConnection(testUtil.getConfiguration());
      Table acl = conn.getTable(AccessControlLists.ACL_TABLE_NAME)) {
    BlockingRpcChannel channel = acl.coprocessorService(new byte[0]);
    AccessControlService.BlockingInterface protocol =
      AccessControlService.newBlockingStub(channel);
    try {
      protocol.checkPermissions(null,request.build());
    } catch (ServiceException se) {
      ProtobufUtil.toIOException(se);
    }
  }
}

@Test (timeout=180000)
public void testGlobalPermissionList() throws Exception {
  List<UserPermission> perms;
  Table acl = systemUserConnection.getTable(AccessControlLists.ACL_TABLE_NAME);
  try {
    BlockingRpcChannel service = acl.coprocessorService(HConstants.EMPTY_START_ROW);
    AccessControlService.BlockingInterface protocol =
      AccessControlService.newBlockingStub(service);
    perms = AccessControlUtil.getUserPermissions(null,protocol);
  } finally {
    acl.close();
  }

  Collection<String> superUsers = Superusers.getSuperUsers();
  List<UserPermission> adminPerms = new ArrayList<>(superUsers.size() + 1);
  adminPerms.add(new UserPermission(Bytes.toBytes(USER_ADMIN.getShortName()),Bytes.toBytes("ACRW")));

  for(String user: superUsers) {
    adminPerms.add(new UserPermission(Bytes.toBytes(user),Action.values()));
  }
  assertTrue("Only super users,global users and user admin has permission on table hbase:acl " +
      "per setup",perms.size() == 5 + superUsers.size() &&
      hasFoundUserPermission(adminPerms,perms));
}

BlockingRpcclient(final Class<?> protocol,final InetSocketAddress addr,ClientSocketChannelFactory factory)
    throws Exception {

  this.protocol = protocol;
  String serviceClassName = protocol.getName() + "$"
      + protocol.getSimpleName() + "Service";
  Class<?> serviceClass = Class.forName(serviceClassName);
  stubMethod = serviceClass.getmethod("newBlockingStub",BlockingRpcChannel.class);

  this.handler = new ClientChannelUpstreamHandler();
  pipeFactory = new ProtoPipelineFactory(handler,RpcResponse.getDefaultInstance());
  super.init(addr,pipeFactory,factory);
  rpcChannel = new ProxyRpcChannel();

  this.key = new RpcConnectionKey(addr,false);
}

BlockingRpcclient(final Class<?> protocol,false);
}

/**
 * Grant permissions globally to the given user. Will wait until all active
 * AccessController instances have updated their permissions caches or will
 * throw an exception upon timeout (10 seconds).
 */
public static void grantGlobal(final hbasetestingutility util,actions);
      } finally {
        acl.close();
      }
      return null;
    }
  });
}

/**
 * Revoke permissions globally from the given user. Will wait until all active
 * AccessController instances have updated their permissions caches or will
 * throw an exception upon timeout (10 seconds).
 */
public static void revokeGlobal(final hbasetestingutility util,actions);
      } finally {
        acl.close();
      }
      return null;
    }
  });
}

private EnumOptions generateEnumOptions(Enum e) {
  EnumOptions.Builder builder = EnumOptions.newBuilder();
  setoptions(builder,e.getoptionsList(),ENUM_OPTION_NAME_PREFIX);
  return builder.build();
}

/** Get the {@code EnumOptions},defined in {@code descriptor.proto}. */
public EnumOptions getoptions() { return proto.getoptions(); }

/**
 * Get the {@code EnumOptions},defined in {@code descriptor.proto}.
 */
public EnumOptions getoptions () {
    return proto.getoptions ();
}

public EnumOptions.Builder getEnumOptions() {
  return currentScope.getEnumOptions();
}

protected EnumOptions.Builder getEnumOptions() {
  throw new RuntimeException(NOT_APPLICABLE_IN_CURRENT_ScopE);
}

@Override
protected EnumOptions.Builder getEnumOptions() {
  return protoBuilder.getoptionsBuilder();
}