DatastoreStorageTransaction implements StorageTransaction { @Override public void store(Shard shard) throws IOException { tx.put(Entity.newBuilder(tx.getDatastore().newKeyFactory().setKind(KIND_COUNTER_SHARD) .newKey(shard.counterId() + "-" + shard.index())) .set(PROPERTY_COUNTER_ID, shard.counterId()) .set(PROPERTY_SHARD_INDEX, shard.index()) .set(PROPERTY_SHARD_VALUE, shard.value()) .build()); } DatastoreStorageTransaction(CheckedDatastoreTransaction transaction); @Override void updateCounter(ShardedCounter shardedCounter, String resource, int delta); @Override Optional<Shard> shard(String counterId, int shardIndex); @Override void store(Shard shard); @Override void updateLimitForCounter(String counterId, long limit); @Override void store(Resource resource); @Override WorkflowId deleteWorkflow(WorkflowId workflowId); @Override WorkflowId store(Workflow workflow); @Override WorkflowId storeWorkflowWithNextNaturalTrigger(Workflow workflow, TriggerInstantSpec triggerSpec); @Override Optional<Workflow> workflow(WorkflowId workflowId); @Override WorkflowId updateNextNaturalTrigger(WorkflowId workflowId, TriggerInstantSpec triggerSpec); @Override WorkflowId patchState(WorkflowId workflowId, WorkflowState state); @Override Optional<RunState> readActiveState(WorkflowInstance instance); @Override WorkflowInstance writeActiveState(WorkflowInstance instance, RunState state); @Override WorkflowInstance updateActiveState(WorkflowInstance instance, RunState state); @Override WorkflowInstance deleteActiveState(WorkflowInstance instance); @Override Backfill store(Backfill backfill); @Override Optional<Backfill> backfill(String id);  }
@Test public void shouldStoreShards() throws IOException { storage.runInTransactionWithRetries(tx -> { Shard shard1 = Shard.create("res1", 0, 1); Shard shard2 = Shard.create("res1", 1, 1); tx.store(shard1); tx.store(shard2); return null; }); assertThat(storage.shardsForCounter("res1").size(), is(2)); }