@Test public void testMultipleFields() throws Exception { GraphQLSchema schema = newQuerySchema(it -> it .field(newLongField("a").dataFetcher(env -> Flowable.interval(300, MILLISECONDS, scheduler).take(8))) .field(newLongField("b").dataFetcher(env -> Flowable.timer(2, SECONDS, scheduler))) .field(newLongField("c").dataFetcher(env -> Flowable.just(42L))) ); ExecutionResult executionResult = new GraphQL(schema, strategy).execute("{ a, b, c }"); Flowable.fromPublisher((Publisher<Change>) executionResult.getData()).timestamp(scheduler).subscribe(subscriber); subscriber.assertEmpty(); scheduler.advanceTimeBy(1, SECONDS); subscriber.assertEmpty(); scheduler.advanceTimeBy(2, SECONDS); subscriber .assertChanges(it -> it.containsExactly( tuple("02:000", "", ImmutableMap.of("a", 5L, "b", 0L, "c", 42L)), tuple("02:100", "a", 6L), tuple("02:400", "a", 7L) )) .assertComplete(); }
@Override public ExecutionResult execute(ExecutionContext context, ExecutionParameters parameters) throws NonNullableFieldWasNullException { Map<String, List<Field>> fields = parameters.fields(); Object source = parameters.source(); return complexChangesFlow( context, source, __ -> fields.entrySet().stream(), (entry, sourceValue) -> resolveField( new ReactiveContext(context, entry.getKey()), transformParametersSource(parameters, source == null ? null : sourceValue), entry.getValue() ), results -> { Map<Object, Object> result = new HashMap<>(); for (Object entry : results) { result.put(((SimpleEntry) entry).getKey(), ((SimpleEntry) entry).getValue()); } return result; } ); }
ReactiveExecutionStrategy extends ExecutionStrategy { @Override public ExecutionResult execute(ExecutionContext context, ExecutionParameters parameters) throws NonNullableFieldWasNullException { Map<String, List<Field>> fields = parameters.fields(); Object source = parameters.source(); return complexChangesFlow( context, source, __ -> fields.entrySet().stream(), (entry, sourceValue) -> resolveField( new ReactiveContext(context, entry.getKey()), transformParametersSource(parameters, source == null ? null : sourceValue), entry.getValue() ), results -> { Map<Object, Object> result = new HashMap<>(); for (Object entry : results) { result.put(((SimpleEntry) entry).getKey(), ((SimpleEntry) entry).getValue()); } return result; } ); } }
ReactiveExecutionStrategy extends ExecutionStrategy { @Override public ExecutionResult execute(ExecutionContext context, ExecutionParameters parameters) throws NonNullableFieldWasNullException { Map<String, List<Field>> fields = parameters.fields(); Object source = parameters.source(); return complexChangesFlow( context, source, __ -> fields.entrySet().stream(), (entry, sourceValue) -> resolveField( new ReactiveContext(context, entry.getKey()), transformParametersSource(parameters, source == null ? null : sourceValue), entry.getValue() ), results -> { Map<Object, Object> result = new HashMap<>(); for (Object entry : results) { result.put(((SimpleEntry) entry).getKey(), ((SimpleEntry) entry).getValue()); } return result; } ); }  }
ReactiveExecutionStrategy extends ExecutionStrategy { @Override public ExecutionResult execute(ExecutionContext context, ExecutionParameters parameters) throws NonNullableFieldWasNullException { Map<String, List<Field>> fields = parameters.fields(); Object source = parameters.source(); return complexChangesFlow( context, source, __ -> fields.entrySet().stream(), (entry, sourceValue) -> resolveField( new ReactiveContext(context, entry.getKey()), transformParametersSource(parameters, source == null ? null : sourceValue), entry.getValue() ), results -> { Map<Object, Object> result = new HashMap<>(); for (Object entry : results) { result.put(((SimpleEntry) entry).getKey(), ((SimpleEntry) entry).getValue()); } return result; } ); }  @Override ExecutionResult execute(ExecutionContext context, ExecutionParameters parameters); }
ReactiveExecutionStrategy extends ExecutionStrategy { @Override public ExecutionResult execute(ExecutionContext context, ExecutionParameters parameters) throws NonNullableFieldWasNullException { Map<String, List<Field>> fields = parameters.fields(); Object source = parameters.source(); return complexChangesFlow( context, source, __ -> fields.entrySet().stream(), (entry, sourceValue) -> resolveField( new ReactiveContext(context, entry.getKey()), transformParametersSource(parameters, source == null ? null : sourceValue), entry.getValue() ), results -> { Map<Object, Object> result = new HashMap<>(); for (Object entry : results) { result.put(((SimpleEntry) entry).getKey(), ((SimpleEntry) entry).getValue()); } return result; } ); }  @Override ExecutionResult execute(ExecutionContext context, ExecutionParameters parameters);  }