Skip to content

Commit

Permalink
Clean up api for group by var args key fields.
Browse files Browse the repository at this point in the history
  • Loading branch information
@greg-higgins
greg-higgins committed Dec 3, 2023
1 parent c63671c commit 71e5e58
Show file tree
Hide file tree
Showing 2 changed files with 13 additions and 15 deletions.
16 changes: 4 additions & 12 deletions compiler/src/main/java/com/fluxtion/compiler/builder/dataflow/FlowBuilder.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -218,23 +218,20 @@ public <K> GroupByFlowBuilder<K, T> groupBy(SerializableFunction<T, K> keyFuncti
* Creates a GroupByFlowBuilder using a compound key created by a set of method reference accessors to for the value.
* The value is the last value supplied
*
* @param keyFunction key accessor
* @param keyFunctions multi arg key accessors
* @return GroupByFlowBuilder keyed on properties
*/
@SafeVarargs
public final GroupByFlowBuilder<GroupByKey<T>, T> groupByFields(
SerializableFunction<T, ?> keyFunction,
SerializableFunction<T, ?>... keyFunctions) {
return groupBy(GroupByKey.build(keyFunction, keyFunctions));
return groupBy(GroupByKey.build(keyFunctions));
}

/**
* Aggregates a flow using a key to group by and an aggregating function to process new values for a keyed
* bucket. The key is a compound key created by a set of method reference accessors to for the value.
*
* @param aggregateFunctionSupplier A factory that supplies aggregating functions, each function has its own function instance
* @param keyFunction key accessor
* @param keyFunctions multi arg key accessors
* @param <A> The return type of the aggregating function
* @param <F> The aggregating function type
Expand All @@ -244,26 +241,23 @@ public final GroupByFlowBuilder<GroupByKey<T>, T> groupByFields(
@SafeVarargs
public final <A, F extends AggregateFlowFunction<T, A, F>> GroupByFlowBuilder<GroupByKey<T>, A> groupByFieldsAggregate(
SerializableSupplier<F> aggregateFunctionSupplier,
SerializableFunction<T, ?> keyFunction,
SerializableFunction<T, ?>... keyFunctions) {
return groupBy(GroupByKey.build(keyFunction, keyFunctions), aggregateFunctionSupplier);
return groupBy(GroupByKey.build(keyFunctions), aggregateFunctionSupplier);
}

/**
* Creates a GroupByFlowBuilder using a compound key created by a set of method reference accessors to for the key
* The value is extracted from the input using the value function
*
* @param valueFunction the value that will be stored in the groupBy
* @param keyFunction key accessor
* @param keyFunctions multi arg key accessors
* @return GroupByFlowBuilder keyed on properties
*/
@SafeVarargs
public final <V> GroupByFlowBuilder<GroupByKey<T>, V> groupByFieldsAndGet(
SerializableFunction<T, V> valueFunction,
SerializableFunction<T, ?> keyFunction,
SerializableFunction<T, ?>... keyFunctions) {
return groupBy(GroupByKey.build(keyFunction, keyFunctions), valueFunction);
return groupBy(GroupByKey.build(keyFunctions), valueFunction);
}

/**
Expand All @@ -272,7 +266,6 @@ public final <V> GroupByFlowBuilder<GroupByKey<T>, V> groupByFieldsAndGet(
*
* @param valueFunction the value that will be stored in the groupBy
* @param aggregateFunctionSupplier A factory that supplies aggregating functions, each function has its own function instance
* @param keyFunction key accessor
* @param keyFunctions multi arg key accessors
* @param <V> Value type extracted from the incoming data flow
* @param <A> The return type of the aggregating function
Expand All @@ -284,9 +277,8 @@ public final <V> GroupByFlowBuilder<GroupByKey<T>, V> groupByFieldsAndGet(
public final <V, A, F extends AggregateFlowFunction<V, A, F>> GroupByFlowBuilder<GroupByKey<T>, A> groupByFieldsGetAndAggregate(
SerializableFunction<T, V> valueFunction,
SerializableSupplier<F> aggregateFunctionSupplier,
SerializableFunction<T, ?> keyFunction,
SerializableFunction<T, ?>... keyFunctions) {
return groupBy(GroupByKey.build(keyFunction, keyFunctions), valueFunction, aggregateFunctionSupplier);
return groupBy(GroupByKey.build(keyFunctions), valueFunction, aggregateFunctionSupplier);
}

public <K> GroupByFlowBuilder<K, List<T>> groupByToList(SerializableFunction<T, K> keyFunction) {
Expand Down
12 changes: 9 additions & 3 deletions runtime/src/main/java/com/fluxtion/runtime/dataflow/groupby/GroupByKey.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
import lombok.Getter;
import lombok.ToString;

import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
Expand All @@ -18,20 +19,25 @@
@ToString(of = {"key", "name"})
public class GroupByKey<T> {
public final List<LambdaReflection.SerializableFunction<T, ?>> accessors;
public final transient List<Method> accessorsMethods;
private final transient StringBuilder keyHolder = new StringBuilder();
@Getter
private final transient Class<T> valueClass;
@Getter
private transient String key;
@Getter
private transient T keyInstance;
private transient final String name;

public GroupByKey(List<LambdaReflection.SerializableFunction<T, ?>> accessorsToAdd) {
this.accessors = new ArrayList<>();
this.accessorsMethods = new ArrayList<>();
String tmpName = "";
for (LambdaReflection.SerializableFunction<T, ?> element : accessorsToAdd) {
if (!accessors.contains(element)) {
accessors.add(element);
tmpName += "_" + element.method().getName();
accessorsMethods.add(element.method());
}
}
valueClass = (Class<T>) accessors.get(0).method().getDeclaringClass();
Expand All @@ -49,6 +55,7 @@ public GroupByKey(LambdaReflection.SerializableFunction<T, ?>... accessorList) {

private GroupByKey(GroupByKey<T> toClone) {
accessors = toClone.accessors;
accessorsMethods = toClone.accessorsMethods;
valueClass = toClone.getValueClass();
name = toClone.name;
}
Expand All @@ -59,24 +66,23 @@ public static <T> LambdaReflection.SerializableFunction<T, GroupByKey<T>> build(

@SafeVarargs
public static <T> LambdaReflection.SerializableFunction<T, GroupByKey<T>> build(
LambdaReflection.SerializableFunction<T, ?> accessor,
LambdaReflection.SerializableFunction<T, ?>... accessorList) {
List<LambdaReflection.SerializableFunction<T, ?>> accessors = new ArrayList<>();
accessors.add(accessor);
accessors.addAll(Arrays.asList(accessorList));
GroupByKey<T> accessorKey = new GroupByKey<>(accessors);
return accessorKey::toKey;
}


public boolean keyPresent(LambdaReflection.SerializableFunction<T, ?> keyToCheck) {
return accessors.contains(keyToCheck);
return accessorsMethods.contains(keyToCheck.method());
}

public GroupByKey<T> toKey(T input) {
//TODO add object pooling
GroupByKey<T> cloned = new GroupByKey<>(this);
cloned.keyHolder.setLength(0);
cloned.keyInstance = input;
for (int i = 0, accessorsSize = accessors.size(); i < accessorsSize; i++) {
LambdaReflection.SerializableFunction<T, ?> accessor = accessors.get(i);
cloned.keyHolder.append(accessor.apply(input).toString());
Expand Down

0 comments on commit 71e5e58

Please sign in to comment.