ShardingService { public List<Integer> getShardingItems(final String jobInstanceId) { JobInstance jobInstance = new JobInstance(jobInstanceId); if (!serverService.isAvailableServer(jobInstance.getIp())) { return Collections.emptyList(); } List<Integer> result = new LinkedList<>(); int shardingTotalCount = configService.load(true).getShardingTotalCount(); for (int i = 0; i < shardingTotalCount; i++) { if (jobInstance.getJobInstanceId().equals(jobNodeStorage.getJobNodeData(ShardingNode.getInstanceNode(i)))) { result.add(i); } } return result; } ShardingService(final CoordinatorRegistryCenter regCenter, final String jobName); void setReshardingFlag(); boolean isNeedSharding(); void shardingIfNecessary(); List<Integer> getShardingItems(final String jobInstanceId); List<Integer> getLocalShardingItems(); boolean hasShardingInfoInOfflineServers();  }
@Test public void assertGetShardingItemsWithNotAvailableServer() { assertThat(shardingService.getShardingItems("127.0.0.1@-@0"), is(Collections.<Integer>emptyList())); }
@Test public void assertGetShardingItemsWithEnabledServer() { JobRegistry.getInstance().registerRegistryCenter("test_job", regCenter); JobRegistry.getInstance().registerJob("test_job", jobScheduleController); when(serverService.isAvailableServer("127.0.0.1")).thenReturn(true); when(configService.load(true)).thenReturn(JobConfiguration.newBuilder("test_job", 3).cron("0/1 * * * * ?").build()); when(jobNodeStorage.getJobNodeData("sharding/0/instance")).thenReturn("127.0.0.1@-@0"); when(jobNodeStorage.getJobNodeData("sharding/1/instance")).thenReturn("127.0.0.1@-@1"); when(jobNodeStorage.getJobNodeData("sharding/2/instance")).thenReturn("127.0.0.1@-@0"); assertThat(shardingService.getShardingItems("127.0.0.1@-@0"), is(Arrays.asList(0, 2))); JobRegistry.getInstance().shutdown("test_job"); }