@Test public void testGetAssignmentFromStreamSingleStateNotEmpty() { List<Assignment.Builder> mockedResult = new ArrayList<>(); mockedResult.add(Assignment.newInstance(Experiment.ID.valueOf(experimentId)) .withBucketLabel(Bucket.Label.valueOf("label1")) ); Bucket mockedBucket = Bucket.newInstance(Experiment.ID.valueOf(experimentId), Bucket.Label.valueOf("label1")) .withAllocationPercent(1.0) .withState(Bucket.State.OPEN) .build(); when(experimentRepository.getBucket(eq(Experiment.ID.valueOf(experimentId)), eq(Bucket.Label.valueOf("label1")))) .thenReturn(mockedBucket); Optional<Assignment> assignmentOptional = repository.getAssignmentFromStream( Experiment.ID.valueOf(experimentId), User.ID.valueOf("testuser1"), Context.valueOf("test"), mockedResult.stream()); assertThat(assignmentOptional.isPresent(), is(true)); assertThat(assignmentOptional.get().getBucketLabel().toString(), is("label1")); assertThat(assignmentOptional.get().isCacheable(), is(false)); assertThat(assignmentOptional.get().isBucketEmpty(), is(false)); }
Optional<Assignment> getAssignmentFromStream(Experiment.ID experimentID, User.ID userID, Context context, Stream<Assignment.Builder> resultStream) { return resultStream.map(t -> { Bucket.Label bucketLabel = t.getBucketLabel(); boolean isBucketEmpty = false; if (bucketLabel != null && experimentRepository.getBucket(experimentID, t.getBucketLabel()).getState().equals(Bucket.State.EMPTY)) { bucketLabel = null; isBucketEmpty = true; } t.withStatus(Assignment.Status.EXISTING_ASSIGNMENT) .withCacheable(false) .withBucketEmpty(isBucketEmpty) .withBucketLabel(bucketLabel); return t.build(); }).reduce((element, anotherElement) -> { throw new RepositoryException("Multiple element fetched from db for experimentId = \"" + experimentID + "\" userID = \"" + userID + " context=\"" + context.getContext() + "\""); }); }
CassandraAssignmentsRepository implements AssignmentsRepository { Optional<Assignment> getAssignmentFromStream(Experiment.ID experimentID, User.ID userID, Context context, Stream<Assignment.Builder> resultStream) { return resultStream.map(t -> { Bucket.Label bucketLabel = t.getBucketLabel(); boolean isBucketEmpty = false; if (bucketLabel != null && experimentRepository.getBucket(experimentID, t.getBucketLabel()).getState().equals(Bucket.State.EMPTY)) { bucketLabel = null; isBucketEmpty = true; } t.withStatus(Assignment.Status.EXISTING_ASSIGNMENT) .withCacheable(false) .withBucketEmpty(isBucketEmpty) .withBucketLabel(bucketLabel); return t.build(); }).reduce((element, anotherElement) -> { throw new RepositoryException("Multiple element fetched from db for experimentId = \"" + experimentID + "\" userID = \"" + userID + " context=\"" + context.getContext() + "\""); }); } }
CassandraAssignmentsRepository implements AssignmentsRepository { Optional<Assignment> getAssignmentFromStream(Experiment.ID experimentID, User.ID userID, Context context, Stream<Assignment.Builder> resultStream) { return resultStream.map(t -> { Bucket.Label bucketLabel = t.getBucketLabel(); boolean isBucketEmpty = false; if (bucketLabel != null && experimentRepository.getBucket(experimentID, t.getBucketLabel()).getState().equals(Bucket.State.EMPTY)) { bucketLabel = null; isBucketEmpty = true; } t.withStatus(Assignment.Status.EXISTING_ASSIGNMENT) .withCacheable(false) .withBucketEmpty(isBucketEmpty) .withBucketLabel(bucketLabel); return t.build(); }).reduce((element, anotherElement) -> { throw new RepositoryException("Multiple element fetched from db for experimentId = \"" + experimentID + "\" userID = \"" + userID + " context=\"" + context.getContext() + "\""); }); } @Inject  CassandraAssignmentsRepository(
            @CassandraRepository ExperimentRepository experimentRepository,
            @DatabaseRepository ExperimentRepository dbRepository,
            EventLog eventLog,
            ExperimentAccessor experimentAccessor,
            ExperimentUserIndexAccessor experimentUserIndexAccessor,

            UserAssignmentExportAccessor userAssignmentExportAccessor,

            BucketAccessor bucketAccessor,
            BucketAssignmentCountAccessor bucketAssignmentCountAccessor,

            StagingAccessor stagingAccessor,
            PrioritiesAccessor prioritiesAccessor,
            ExclusionAccessor exclusionAccessor,
            PageExperimentIndexAccessor pageExperimentIndexAccessor,
            CassandraDriver driver,
            MappingManager mappingManager,
            @Named("AssignmentsCountThreadPoolExecutor") ThreadPoolExecutor assignmentsCountExecutor,
            final @Named("assign.user.to.export") boolean assignUserToExport,
            final @Named("assign.bucket.count") boolean assignBucketCount,
            final @Named("default.time.format") String defaultTimeFormat); }
CassandraAssignmentsRepository implements AssignmentsRepository { Optional<Assignment> getAssignmentFromStream(Experiment.ID experimentID, User.ID userID, Context context, Stream<Assignment.Builder> resultStream) { return resultStream.map(t -> { Bucket.Label bucketLabel = t.getBucketLabel(); boolean isBucketEmpty = false; if (bucketLabel != null && experimentRepository.getBucket(experimentID, t.getBucketLabel()).getState().equals(Bucket.State.EMPTY)) { bucketLabel = null; isBucketEmpty = true; } t.withStatus(Assignment.Status.EXISTING_ASSIGNMENT) .withCacheable(false) .withBucketEmpty(isBucketEmpty) .withBucketLabel(bucketLabel); return t.build(); }).reduce((element, anotherElement) -> { throw new RepositoryException("Multiple element fetched from db for experimentId = \"" + experimentID + "\" userID = \"" + userID + " context=\"" + context.getContext() + "\""); }); } @Inject  CassandraAssignmentsRepository(
            @CassandraRepository ExperimentRepository experimentRepository,
            @DatabaseRepository ExperimentRepository dbRepository,
            EventLog eventLog,
            ExperimentAccessor experimentAccessor,
            ExperimentUserIndexAccessor experimentUserIndexAccessor,

            UserAssignmentExportAccessor userAssignmentExportAccessor,

            BucketAccessor bucketAccessor,
            BucketAssignmentCountAccessor bucketAssignmentCountAccessor,

            StagingAccessor stagingAccessor,
            PrioritiesAccessor prioritiesAccessor,
            ExclusionAccessor exclusionAccessor,
            PageExperimentIndexAccessor pageExperimentIndexAccessor,
            CassandraDriver driver,
            MappingManager mappingManager,
            @Named("AssignmentsCountThreadPoolExecutor") ThreadPoolExecutor assignmentsCountExecutor,
            final @Named("assign.user.to.export") boolean assignUserToExport,
            final @Named("assign.bucket.count") boolean assignBucketCount,
            final @Named("default.time.format") String defaultTimeFormat); @Override @Timed void populateAssignmentsMetadata(User.ID userID, Application.Name appName, Context context, ExperimentBatch experimentBatch, Optional<Map<Experiment.ID, Boolean>> allowAssignments,
                                            PrioritizedExperimentList prioritizedExperimentList,
                                            Map<Experiment.ID, com.intuit.wasabi.experimentobjects.Experiment> experimentMap,
                                            Map<Experiment.ID, BucketList> bucketMap,
                                            Map<Experiment.ID, List<Experiment.ID>> exclusionMap
    ); @Override @Timed List<Pair<Experiment, String>> getAssignments(User.ID userID,
                                                         Application.Name appLabel,
                                                         Context context,
                                                         Map<Experiment.ID, Experiment> experimentMap); @Override @Timed Assignment getAssignment(User.ID userID, Application.Name appName, Experiment.ID experimentID, Context context); @Override @Timed void assignUsersInBatch(List<Pair<Experiment, Assignment>> assignments, Date date); @Override void assignUserToExports(Assignment assignment, Date date); @Override void deleteAssignment(Experiment experiment, User.ID userID, Context context, Application.Name appName, Assignment currentAssignment); @Override StreamingOutput getAssignmentStream(Experiment.ID experimentID, Context context, Parameters parameters, Boolean ignoreNullBucket); @Override @Timed void pushAssignmentToStaging(String type, String exception, String data); void pushAssignmentsToStaging(String type, String exception, Collection<String> data); Map<UUID, String> getBatchPayloadsFromStaging(int batchSize); void deleteFromStaging(UUID timeUUID); @Override void updateBucketAssignmentCount(Experiment experiment, Assignment assignment, boolean countUp); @Override AssignmentCounts getBucketAssignmentCount(Experiment experiment); @Override Map<Experiment.ID, AssignmentCounts> getBucketAssignmentCountsInParallel(List<Experiment.ID> experimentIds); }
CassandraAssignmentsRepository implements AssignmentsRepository { Optional<Assignment> getAssignmentFromStream(Experiment.ID experimentID, User.ID userID, Context context, Stream<Assignment.Builder> resultStream) { return resultStream.map(t -> { Bucket.Label bucketLabel = t.getBucketLabel(); boolean isBucketEmpty = false; if (bucketLabel != null && experimentRepository.getBucket(experimentID, t.getBucketLabel()).getState().equals(Bucket.State.EMPTY)) { bucketLabel = null; isBucketEmpty = true; } t.withStatus(Assignment.Status.EXISTING_ASSIGNMENT) .withCacheable(false) .withBucketEmpty(isBucketEmpty) .withBucketLabel(bucketLabel); return t.build(); }).reduce((element, anotherElement) -> { throw new RepositoryException("Multiple element fetched from db for experimentId = \"" + experimentID + "\" userID = \"" + userID + " context=\"" + context.getContext() + "\""); }); } @Inject  CassandraAssignmentsRepository(
            @CassandraRepository ExperimentRepository experimentRepository,
            @DatabaseRepository ExperimentRepository dbRepository,
            EventLog eventLog,
            ExperimentAccessor experimentAccessor,
            ExperimentUserIndexAccessor experimentUserIndexAccessor,

            UserAssignmentExportAccessor userAssignmentExportAccessor,

            BucketAccessor bucketAccessor,
            BucketAssignmentCountAccessor bucketAssignmentCountAccessor,

            StagingAccessor stagingAccessor,
            PrioritiesAccessor prioritiesAccessor,
            ExclusionAccessor exclusionAccessor,
            PageExperimentIndexAccessor pageExperimentIndexAccessor,
            CassandraDriver driver,
            MappingManager mappingManager,
            @Named("AssignmentsCountThreadPoolExecutor") ThreadPoolExecutor assignmentsCountExecutor,
            final @Named("assign.user.to.export") boolean assignUserToExport,
            final @Named("assign.bucket.count") boolean assignBucketCount,
            final @Named("default.time.format") String defaultTimeFormat); @Override @Timed void populateAssignmentsMetadata(User.ID userID, Application.Name appName, Context context, ExperimentBatch experimentBatch, Optional<Map<Experiment.ID, Boolean>> allowAssignments,
                                            PrioritizedExperimentList prioritizedExperimentList,
                                            Map<Experiment.ID, com.intuit.wasabi.experimentobjects.Experiment> experimentMap,
                                            Map<Experiment.ID, BucketList> bucketMap,
                                            Map<Experiment.ID, List<Experiment.ID>> exclusionMap
    ); @Override @Timed List<Pair<Experiment, String>> getAssignments(User.ID userID,
                                                         Application.Name appLabel,
                                                         Context context,
                                                         Map<Experiment.ID, Experiment> experimentMap); @Override @Timed Assignment getAssignment(User.ID userID, Application.Name appName, Experiment.ID experimentID, Context context); @Override @Timed void assignUsersInBatch(List<Pair<Experiment, Assignment>> assignments, Date date); @Override void assignUserToExports(Assignment assignment, Date date); @Override void deleteAssignment(Experiment experiment, User.ID userID, Context context, Application.Name appName, Assignment currentAssignment); @Override StreamingOutput getAssignmentStream(Experiment.ID experimentID, Context context, Parameters parameters, Boolean ignoreNullBucket); @Override @Timed void pushAssignmentToStaging(String type, String exception, String data); void pushAssignmentsToStaging(String type, String exception, Collection<String> data); Map<UUID, String> getBatchPayloadsFromStaging(int batchSize); void deleteFromStaging(UUID timeUUID); @Override void updateBucketAssignmentCount(Experiment experiment, Assignment assignment, boolean countUp); @Override AssignmentCounts getBucketAssignmentCount(Experiment experiment); @Override Map<Experiment.ID, AssignmentCounts> getBucketAssignmentCountsInParallel(List<Experiment.ID> experimentIds); static final Bucket.Label NULL_LABEL; }