implememented Distinct

Author: jucardi

streams/global.go

@@ -112,26 +112,6 @@ func NewCollectionFromMap(m interface{}) (ICollection, error) {
 	}).init(), nil
 }
 
-//func NewCollectionFromMap(m interface{}) (ICollection, error) {
-//	// Ideally, a collection implementation from a Map would have been defined that knows how to iterate over a K,V set to avoid iterating over the full map initially generate
-//	// an array of `*KeyValuePair`. However, there is not way to iterate over a K,V set of a map through reflection, instead the only thing available is the function `MapKeys`
-//	// which
-//	val := reflect.ValueOf(m)
-//
-//	if val.Kind() != reflect.Map {
-//		return nil, fmt.Errorf("unable to create a key value set collection, the input value must be a map, %s", val.Kind().String())
-//	}
-//
-//	var array []*KeyValuePair
-//	for _, key := range val.MapKeys() {
-//		array = append(array, &KeyValuePair{
-//			Key:   key.Interface(),
-//			Value: val.MapIndex(key),
-//		})
-//	}
-//
-//	return NewCollectionFromArray(array)
-//}
 // NewArrayCollection Creates a new empty array collection of the given type
 //
 // - elementType:  The element type for the items in the collection to be created.

streams/stream.go

@@ -35,6 +35,9 @@ type IStream interface {
 	//            best combine it with a `SortBy`. Only needs to be provided once per stream.
 	Map(f ConvertFunc, threads ...int) IStream
 
+	// Distinct Returns a stream consisting of the distinct elements
+	Distinct() IStream
+
 	// First Returns the first element of the resulting stream.
 	// Returns nil (or default value if provided) if the resulting stream is empty.
 	First(defaultValue ...interface{}) interface{}

streams/stream_impl.go

@@ -13,6 +13,7 @@ type Stream struct {
 	iterable IIterable
 	filters  []ConditionalFunc
 	sorts    []sortFunc
+	distinct bool
 	threads  int
 }
 
@@ -26,6 +27,19 @@ type sorter struct {
 	sorts []sortFunc
 }
 
+type iAdd interface {
+	Add(item interface{}) error
+}
+
+type mapAdd struct {
+	m reflect.Value
+}
+
+func (m *mapAdd) Add(item interface{}) error {
+	m.m.SetMapIndex(reflect.ValueOf(item), reflect.ValueOf(true))
+	return nil
+}
+
 // SetThreads Sets the amount of go channels to be used for parallel filtering to a maximum of the available CPUs in the
 // host machine. Providing a value <= 0, indicates the maximum amount of available CPUs will be the number that determines
 // the amount of go channels to be used. If order matters, best combine it with a `SortBy`. Only needs to be provided once
@@ -91,6 +105,12 @@ func (s *Stream) Map(f ConvertFunc, threads ...int) IStream {
 	return FromIterable(col)
 }
 
+// Distinct Returns a stream consisting of the distinct elements
+func (s *Stream) Distinct() IStream {
+	s.distinct = true
+	return s
+}
+
 // First Returns the first element of the resulting stream.
 // Returns nil (or default value if provided) if the resulting stream is empty.
 //
@@ -347,17 +367,25 @@ func (s *Stream) parallelProcess(threads int) IIterable {
 }
 
 func (s *Stream) filter(iterable IIterable) IIterable {
-	return s.filterHandler(iterable, 0, iterable.Len())
+	return s.iterHandler(iterable, 0, iterable.Len())
 }
 
-func (s *Stream) filterHandler(iterable IIterable, start, end int) IIterable {
-	if len(s.filters) == 0 {
+func (s *Stream) iterHandler(iterable IIterable, start, end int) IIterable {
+	if len(s.filters) == 0 && !s.distinct {
 		return iterable
 	}
 
+	var adder iAdd
+
 	ret := NewArrayCollection(iterable.ElementType())
 	iterator := iterable.Iterator().Skip(start)
 	i := start
+	adder = ret
+
+	if s.distinct {
+		mType := reflect.MapOf(iterable.ElementType(), reflect.TypeOf(true))
+		adder = &mapAdd{m: reflect.MakeMap(mType)}
+	}
 
 	for x := iterator.Current(); iterator.HasNext() && i < end; x = iterator.Next() {
 		i++
@@ -372,7 +400,14 @@ func (s *Stream) filterHandler(iterable IIterable, start, end int) IIterable {
 		}
 
 		if match {
-			ret.Add(x)
+			_ = adder.Add(x)
+		}
+	}
+
+	if s.distinct {
+		mAdder := adder.(*mapAdd)
+		for _, v := range mAdder.m.MapKeys() {
+			_ = ret.Add(v.Interface())
 		}
 	}
 
@@ -381,7 +416,7 @@ func (s *Stream) filterHandler(iterable IIterable, start, end int) IIterable {
 
 func (s *Stream) parallelProcessHandler(iterable IIterable, threads int) IIterable {
 	worker := func(result chan IIterable, start, end int) {
-		result <- s.filterHandler(iterable, start, end)
+		result <- s.iterHandler(iterable, start, end)
 	}
 
 	ret := NewArrayCollection(iterable.ElementType())
@@ -482,7 +517,6 @@ func getCores(threads ...int) int {
 // TODO:
 //
 // STREAM
-//   Distinct
 //   Reverse
 
 // OPTIONAL ?? or element

streams/stream_test.go

@@ -259,3 +259,22 @@ func TestStream_MapCollection(t *testing.T) {
 	jsonResult, _ := json.MarshalIndent(result, "", " ")
 	assert.JSONEq(t, `[{ "Key": "e", "Value": "111" }, { "Key": "a", "Value": "123" }, { "Key": "b", "Value": "456" }, { "Key": "c", "Value": "789" }]`, string(jsonResult))
 }
+
+func TestStream_Distinct(t *testing.T) {
+	arr := append(testArray, "apple", "banana", "kiwi", "apple", "banana", "apple")
+	result := From(arr).Distinct().ToArray().([]string)
+
+	assert.Equal(t, len(result), len(testArray))
+}
+
+func TestStream_DistinctWithSort(t *testing.T) {
+	arr := append(testArray, "apple", "banana", "kiwi", "apple", "banana", "apple")
+	sortFn := func(a interface{}, b interface{}) int {
+		return strings.Compare(a.(string), b.(string))
+	}
+
+	expected := []string{"apple", "banana", "kiwi", "orange", "peach", "pear", "pineapple", "plum"}
+	sorted := From(arr).OrderBy(sortFn).Distinct().ToArray().([]string)
+
+	assert.Equal(t, expected, sorted)
+}