### Question:
ChangesFilter implements Predicate<Changes> { @NonNull public static Flowable<Changes> applyForTablesAndTags( @NonNull Flowable<Changes> changes, @NonNull Set<String> tables, @NonNull Set<String> tags ) { checkNotNull(tables, "Set of tables can not be null"); checkNotNull(tags, "Set of tags can not be null"); return changes .filter(new ChangesFilter(tables, tags)); } private  ChangesFilter(@Nullable Set<String> tables, @Nullable Set<String> tags); @NonNull static Flowable<Changes> applyForTables(@NonNull Flowable<Changes> changes, @NonNull Set<String> tables); @NonNull static Flowable<Changes> applyForTags(@NonNull Flowable<Changes> changes, @NonNull Set<String> tags); @NonNull static Flowable<Changes> applyForTablesAndTags(
            @NonNull Flowable<Changes> changes,
            @NonNull Set<String> tables,
            @NonNull Set<String> tags
    ); @Override boolean test(@NonNull Changes changes);  }### Answer:
@Test public void applyForTablesAndTags_throwsIfTablesNull() { expectedException.expect(NullPointerException.class); expectedException.expectMessage(equalTo("Set of tables can not be null")); expectedException.expectCause(nullValue(Throwable.class)); ChangesFilter.applyForTablesAndTags(Flowable.<Changes>empty(), null, Collections.<String>emptySet()); }
@Test public void applyForTablesAndTags_throwsIfTagsNull() { expectedException.expect(NullPointerException.class); expectedException.expectMessage(equalTo("Set of tags can not be null")); expectedException.expectCause(nullValue(Throwable.class)); ChangesFilter.applyForTablesAndTags(Flowable.<Changes>empty(), Collections.<String>emptySet(), null); }
@Test public void applyForTablesAndTags_shouldNotifyByTable() { final TestSubscriber<Changes> testSubscriber = new TestSubscriber<Changes>(); ChangesFilter .applyForTablesAndTags( Flowable.just(Changes.newInstance("table1", "another_tag"), Changes.newInstance("table2", "tag2"), Changes.newInstance("table3")), singleton("table1"), singleton("tag1")) .subscribe(testSubscriber); testSubscriber.assertValue(Changes.newInstance("table1", "another_tag")); testSubscriber.dispose(); }
@Test public void applyForTablesAndTags_shouldNotifyByTag() { final TestSubscriber<Changes> testSubscriber = new TestSubscriber<Changes>(); ChangesFilter .applyForTablesAndTags( Flowable.just(Changes.newInstance("another_table", "tag1"), Changes.newInstance("table2", "tag2"), Changes.newInstance("table3")), singleton("table1"), singleton("tag1")) .subscribe(testSubscriber); testSubscriber.assertValue(Changes.newInstance("another_table", "tag1")); testSubscriber.dispose(); }
@Test public void applyForTablesAndTags_shouldSendJustOnceNotificationIfBothTableAndTagAreSatisfy() { final TestSubscriber<Changes> testSubscriber = new TestSubscriber<Changes>(); ChangesFilter .applyForTablesAndTags( Flowable.just(Changes.newInstance("target_table", "target_tag")), singleton("target_table"), singleton("target_tag")) .subscribe(testSubscriber); testSubscriber.assertValueCount(1); testSubscriber.assertValue(Changes.newInstance("target_table", "target_tag")); testSubscriber.dispose(); }