@Test public void testNested() throws Exception { GraphQLObjectType innerType = newObject() .name("Inner") .field(newStringField("b").dataFetcher(env -> Flowable.interval(300, MILLISECONDS, scheduler).take(4).map(i -> i + " " + env.getSource()))) .build(); GraphQLSchema schema = newQuerySchema(it -> it .field(newField("a", innerType).dataFetcher(env -> Flowable.interval(1, SECONDS, scheduler).take(3))) ); ExecutionResult executionResult = new GraphQL(schema, strategy).execute("{ a { b } }"); Flowable.fromPublisher((Publisher<Change>) executionResult.getData()).timestamp(scheduler).subscribe(subscriber); scheduler.advanceTimeBy(5, SECONDS); subscriber .assertChanges(it -> it.containsExactly( tuple("01:300", "", ImmutableMap.of("a", ImmutableMap.of("b", "0 0"))), tuple("01:600", "a.b", "1 0"), tuple("01:900", "a.b", "2 0"), tuple("02:300", "a", ImmutableMap.of("b", "0 1")), tuple("02:600", "a.b", "1 1"), tuple("02:900", "a.b", "2 1"), tuple("03:300", "a", ImmutableMap.of("b", "0 2")), tuple("03:600", "a.b", "1 2"), tuple("03:900", "a.b", "2 2"), tuple("04:200", "a.b", "3 2") )) .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);  }