Feat: Replace old Module with reactive version

ds/reactive/set_impl.go

@@ -56,6 +56,9 @@ func (s *set[ElementType]) Apply(mutations ds.SetMutations[ElementType]) (applie
 	defer s.mutex.Unlock()
 
 	appliedMutations, updateID, registeredCallbacks := s.apply(mutations)
+	if appliedMutations.IsEmpty() {
+		return appliedMutations
+	}
 
 	for _, registeredCallback := range registeredCallbacks {
 		if registeredCallback.LockExecution(updateID) {

ds/reactive/wait_group.go

@@ -0,0 +1,28 @@
+package reactive
+
+// WaitGroup is a reactive Event that waits for a set of elements to be done and that allows to inspect the pending
+// elements.
+type WaitGroup[T comparable] interface {
+	// Event returns the event that is triggered when all elements are done.
+	Event
+
+	// Add adds the given elements to the wait group.
+	Add(elements ...T)
+
+	// Done marks the given elements as done and triggers the wait group if all elements are done.
+	Done(elements ...T)
+
+	// Wait waits until all elements are done.
+	Wait()
+
+	// PendingElements returns the currently pending elements.
+	PendingElements() ReadableSet[T]
+
+	// Debug subscribes to the PendingElements and logs the state of the WaitGroup to the console whenever it changes.
+	Debug(optStringer ...func(T) string) (unsubscribe func())
+}
+
+// NewWaitGroup creates a new WaitGroup.
+func NewWaitGroup[T comparable](elements ...T) WaitGroup[T] {
+	return newWaitGroup(elements...)
+}

ds/reactive/wait_group_impl.go

@@ -0,0 +1,87 @@
+package reactive
+
+import (
+	"fmt"
+	"strings"
+	"sync"
+	"sync/atomic"
+
+	"github.com/iotaledger/hive.go/ds"
+	"github.com/iotaledger/hive.go/lo"
+)
+
+// waitGroup is the default implementation of the WaitGroup interface.
+type waitGroup[T comparable] struct {
+	// Event embeds the Event that is triggered when all elements are done.
+	Event
+
+	// pendingElements contains the currently pending elements.
+	pendingElements Set[T]
+
+	// pendingElementsCounter is the thread-safe counter that keeps track of the number of pending elements.
+	pendingElementsCounter atomic.Int32
+}
+
+// newWaitGroup creates a new wait group.
+func newWaitGroup[T comparable](elements ...T) *waitGroup[T] {
+	w := &waitGroup[T]{
+		Event:           NewEvent(),
+		pendingElements: NewSet[T](),
+	}
+
+	w.Add(elements...)
+
+	return w
+}
+
+// Add adds the given elements to the wait group.
+func (w *waitGroup[T]) Add(elements ...T) {
+	// first increase the counter so that the trigger is not executed before all elements are added
+	w.pendingElementsCounter.Add(int32(len(elements)))
+
+	// then add the elements (and correct the counter if the elements are already present)
+	for _, element := range elements {
+		if !w.pendingElements.Add(element) {
+			w.pendingElementsCounter.Add(-1)
+		}
+	}
+}
+
+// Done marks the given elements as done and triggers the wait group if all elements are done.
+func (w *waitGroup[T]) Done(elements ...T) {
+	for _, element := range elements {
+		if w.pendingElements.Delete(element) && w.pendingElementsCounter.Add(-1) == 0 {
+			w.Trigger()
+		}
+	}
+}
+
+// Wait waits until all elements are done.
+func (w *waitGroup[T]) Wait() {
+	var wg sync.WaitGroup
+
+	wg.Add(1)
+	w.OnTrigger(wg.Done)
+	wg.Wait()
+}
+
+// PendingElements returns the currently pending elements.
+func (w *waitGroup[T]) PendingElements() ReadableSet[T] {
+	return w.pendingElements
+}
+
+// Debug subscribes to the PendingElements and logs the state of the WaitGroup to the console whenever it changes.
+func (w *waitGroup[T]) Debug(optStringer ...func(T) string) (unsubscribe func()) {
+	return w.pendingElements.OnUpdate(func(_ ds.SetMutations[T]) {
+		pendingElementsString := "DONE"
+		if pendingElements := w.pendingElements.ToSlice(); len(pendingElements) != 0 {
+			stringer := lo.First(optStringer, func(element T) string {
+				return fmt.Sprint(element)
+			})
+
+			pendingElementsString = strings.Join(lo.Map(pendingElements, stringer), ", ")
+		}
+
+		fmt.Println("Waiting:", pendingElementsString)
+	})
+}

ds/reactive/wait_group_test.go

@@ -0,0 +1,40 @@
+package reactive
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestWaitGroup(t *testing.T) {
+	wg := NewWaitGroup(1, 2, 3)
+	wg.Debug()
+
+	require.True(t, wg.PendingElements().HasAll(NewSet(1, 2, 3)))
+	require.False(t, wg.WasTriggered())
+
+	wg.Done(1, 2)
+
+	require.True(t, wg.PendingElements().HasAll(NewSet(3)))
+	require.False(t, wg.WasTriggered())
+
+	wg.Done(1, 2)
+
+	require.True(t, wg.PendingElements().HasAll(NewSet(3)))
+	require.False(t, wg.WasTriggered())
+
+	wg.Add(4)
+
+	require.True(t, wg.PendingElements().HasAll(NewSet(3, 4)))
+	require.False(t, wg.WasTriggered())
+
+	wg.Add(4)
+
+	require.True(t, wg.PendingElements().HasAll(NewSet(3, 4)))
+	require.False(t, wg.WasTriggered())
+
+	wg.Done(3, 4)
+
+	require.True(t, wg.PendingElements().IsEmpty())
+	require.True(t, wg.WasTriggered())
+}

runtime/go.mod

@@ -4,12 +4,12 @@ go 1.22
 
 require (
 	github.com/fjl/memsize v0.0.2
-	github.com/iotaledger/hive.go/constraints v0.0.0-20240223134921-02b9a2022431
-	github.com/iotaledger/hive.go/ds v0.0.0-20240223134921-02b9a2022431
-	github.com/iotaledger/hive.go/ierrors v0.0.0-20240223134921-02b9a2022431
-	github.com/iotaledger/hive.go/lo v0.0.0-20240223134921-02b9a2022431
+	github.com/iotaledger/hive.go/constraints v0.0.0-20240308073758-9a05958cbc65
+	github.com/iotaledger/hive.go/ds v0.0.0-20240311213815-82815d826e6e
+	github.com/iotaledger/hive.go/ierrors v0.0.0-20240308073758-9a05958cbc65
+	github.com/iotaledger/hive.go/lo v0.0.0-20240308073758-9a05958cbc65
 	github.com/iotaledger/hive.go/log v0.0.0-20240223134921-02b9a2022431
-	github.com/iotaledger/hive.go/stringify v0.0.0-20240223134921-02b9a2022431
+	github.com/iotaledger/hive.go/stringify v0.0.0-20240223135123-dc9551f4c690
 	github.com/pelletier/go-toml/v2 v2.1.1
 	github.com/sasha-s/go-deadlock v0.3.1
 	github.com/stretchr/testify v1.8.4
@@ -18,10 +18,10 @@ require (
 
 require (
 	github.com/davecgh/go-spew v1.1.1 // indirect
-	github.com/ethereum/go-ethereum v1.13.13 // indirect
+	github.com/ethereum/go-ethereum v1.13.14 // indirect
 	github.com/holiman/uint256 v1.2.4 // indirect
 	github.com/iancoleman/orderedmap v0.3.0 // indirect
-	github.com/iotaledger/hive.go/serializer/v2 v2.0.0-rc.1.0.20240223134921-02b9a2022431 // indirect
+	github.com/iotaledger/hive.go/serializer/v2 v2.0.0-rc.1.0.20240223135123-dc9551f4c690 // indirect
 	github.com/kr/text v0.2.0 // indirect
 	github.com/petermattis/goid v0.0.0-20231207134359-e60b3f734c67 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect

runtime/go.sum

@@ -6,28 +6,32 @@ github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/decred/dcrd/dcrec/secp256k1/v4 v4.1.0 h1:HbphB4TFFXpv7MNrT52FGrrgVXF1owhMVTHFZIlnvd4=
 github.com/decred/dcrd/dcrec/secp256k1/v4 v4.1.0/go.mod h1:DZGJHZMqrU4JJqFAWUS2UO1+lbSKsdiOoYi9Zzey7Fc=
-github.com/ethereum/go-ethereum v1.13.13 h1:KYn9w7pEWRI9oyZOzO94OVbctSusPByHdFDPj634jII=
-github.com/ethereum/go-ethereum v1.13.13/go.mod h1:TN8ZiHrdJwSe8Cb6x+p0hs5CxhJZPbqB7hHkaUXcmIU=
+github.com/ethereum/go-ethereum v1.13.14 h1:EwiY3FZP94derMCIam1iW4HFVrSgIcpsu0HwTQtm6CQ=
+github.com/ethereum/go-ethereum v1.13.14/go.mod h1:TN8ZiHrdJwSe8Cb6x+p0hs5CxhJZPbqB7hHkaUXcmIU=
 github.com/fjl/memsize v0.0.2 h1:27txuSD9or+NZlnOWdKUxeBzTAUkWCVh+4Gf2dWFOzA=
 github.com/fjl/memsize v0.0.2/go.mod h1:VvhXpOYNQvB+uIk2RvXzuaQtkQJzzIx6lSBe1xv7hi0=
 github.com/holiman/uint256 v1.2.4 h1:jUc4Nk8fm9jZabQuqr2JzednajVmBpC+oiTiXZJEApU=
 github.com/holiman/uint256 v1.2.4/go.mod h1:EOMSn4q6Nyt9P6efbI3bueV4e1b3dGlUCXeiRV4ng7E=
 github.com/iancoleman/orderedmap v0.3.0 h1:5cbR2grmZR/DiVt+VJopEhtVs9YGInGIxAoMJn+Ichc=
 github.com/iancoleman/orderedmap v0.3.0/go.mod h1:XuLcCUkdL5owUCQeF2Ue9uuw1EptkJDkXXS7VoV7XGE=
-github.com/iotaledger/hive.go/constraints v0.0.0-20240223134921-02b9a2022431 h1:8iMRrhjsO4N8f4gojdf1GrclHvX4iv7BmpW28dYHWFM=
-github.com/iotaledger/hive.go/constraints v0.0.0-20240223134921-02b9a2022431/go.mod h1:JF7jjkL6tSUOXm23SWadBzBrl7eJk1DQRLc/fNoVZ+o=
-github.com/iotaledger/hive.go/ds v0.0.0-20240223134921-02b9a2022431 h1:ijIZmcOCVFPizR+ME8ADUkTKdDGh0JF9iCPDGl0Pspk=
-github.com/iotaledger/hive.go/ds v0.0.0-20240223134921-02b9a2022431/go.mod h1:K9LcyfMlvjU5g97aHFOsT7wzrnLnicItwNzRhyCUXag=
-github.com/iotaledger/hive.go/ierrors v0.0.0-20240223134921-02b9a2022431 h1:NT5CpBj1nNnho6UxsaVZl2SFMGUxEuhm9452cKLT2W0=
-github.com/iotaledger/hive.go/ierrors v0.0.0-20240223134921-02b9a2022431/go.mod h1:GQY0/35sjgT9Poi1Vrs9kFVvAkuKzGXfVh4j6CBXsAA=
-github.com/iotaledger/hive.go/lo v0.0.0-20240223134921-02b9a2022431 h1:Qjb97LMqK0NmbSfieT9R5568bpEOsnKtrBrTXoYBDsE=
-github.com/iotaledger/hive.go/lo v0.0.0-20240223134921-02b9a2022431/go.mod h1:67oLzWYiBLGt5PN7IBVHdbt9P6oBYCx9UvMEL8ExDAc=
+github.com/iotaledger/hive.go/constraints v0.0.0-20240308073758-9a05958cbc65 h1:NoYVEDXB0Vd0myF5fEafOwD8nwrbwlAeB6Hm4kFP+0Y=
+github.com/iotaledger/hive.go/constraints v0.0.0-20240308073758-9a05958cbc65/go.mod h1:JF7jjkL6tSUOXm23SWadBzBrl7eJk1DQRLc/fNoVZ+o=
+github.com/iotaledger/hive.go/ds v0.0.0-20240311211056-e4e0a4e04b21 h1:ymrvH4/x6IDqaXjLITOWn5mWmTLilHSPM8A6pk/dGCs=
+github.com/iotaledger/hive.go/ds v0.0.0-20240311211056-e4e0a4e04b21/go.mod h1:wfjeJj9B+MM/3yeUHfvT8Gj8bRsdl9utyh2dZg+1+B0=
+github.com/iotaledger/hive.go/ds v0.0.0-20240311213645-cf91c0c76f7a h1:yQzIeAoZYMZM9/8wMaskchv1jZcQun0Xbn89K766ZMQ=
+github.com/iotaledger/hive.go/ds v0.0.0-20240311213645-cf91c0c76f7a/go.mod h1:wfjeJj9B+MM/3yeUHfvT8Gj8bRsdl9utyh2dZg+1+B0=
+github.com/iotaledger/hive.go/ds v0.0.0-20240311213815-82815d826e6e h1:RUf/NPASqOoLSqMLgWlXFqU+sFSAHNsBu5suzR4PUZs=
+github.com/iotaledger/hive.go/ds v0.0.0-20240311213815-82815d826e6e/go.mod h1:wfjeJj9B+MM/3yeUHfvT8Gj8bRsdl9utyh2dZg+1+B0=
+github.com/iotaledger/hive.go/ierrors v0.0.0-20240308073758-9a05958cbc65 h1:ErLGqfzpzNGFTiHyZEaDraWHCc1pahMCc3EkAhf5Sbo=
+github.com/iotaledger/hive.go/ierrors v0.0.0-20240308073758-9a05958cbc65/go.mod h1:GQY0/35sjgT9Poi1Vrs9kFVvAkuKzGXfVh4j6CBXsAA=
+github.com/iotaledger/hive.go/lo v0.0.0-20240308073758-9a05958cbc65 h1:d2dA8oPzzmr/xX+lO9g2yWhoVclEZApPFwbKJPccPV8=
+github.com/iotaledger/hive.go/lo v0.0.0-20240308073758-9a05958cbc65/go.mod h1:67oLzWYiBLGt5PN7IBVHdbt9P6oBYCx9UvMEL8ExDAc=
 github.com/iotaledger/hive.go/log v0.0.0-20240223134921-02b9a2022431 h1:CxoorcruX5zlz+I3iwPxdfs1D3DJt/NQ60Kb0+YvEKs=
 github.com/iotaledger/hive.go/log v0.0.0-20240223134921-02b9a2022431/go.mod h1:H5tmswUbT3o5+QiM6UPtBv7VnPf+lJtlantgpp2lzUI=
-github.com/iotaledger/hive.go/serializer/v2 v2.0.0-rc.1.0.20240223134921-02b9a2022431 h1:iK9nAxlhvodQbBZwLcrDVCNdajXuvfSzDZuw4HBABwI=
-github.com/iotaledger/hive.go/serializer/v2 v2.0.0-rc.1.0.20240223134921-02b9a2022431/go.mod h1:NK05G4PxwZF1m4jGANJWLhAQ2hP1Nt0L8mgCTFLsSCw=
-github.com/iotaledger/hive.go/stringify v0.0.0-20240223134921-02b9a2022431 h1:x1RySBsaNS59ezNmHZz1rLoGQ8wEr7pivwABUsFzvnQ=
-github.com/iotaledger/hive.go/stringify v0.0.0-20240223134921-02b9a2022431/go.mod h1:O4p7UmsfoeLqtAUwrKbq0lXMxjY/MLQSpZSavvvvGig=
+github.com/iotaledger/hive.go/serializer/v2 v2.0.0-rc.1.0.20240223135123-dc9551f4c690 h1:0ILxOm9L2TjPm/pgCPOOAVqKuBNnVtrNBnHZ+piD6PM=
+github.com/iotaledger/hive.go/serializer/v2 v2.0.0-rc.1.0.20240223135123-dc9551f4c690/go.mod h1:NK05G4PxwZF1m4jGANJWLhAQ2hP1Nt0L8mgCTFLsSCw=
+github.com/iotaledger/hive.go/stringify v0.0.0-20240223135123-dc9551f4c690 h1:whxxnbrtL1h/MJI391wOyPVumb+CO4w+VPgkrQwFYkQ=
+github.com/iotaledger/hive.go/stringify v0.0.0-20240223135123-dc9551f4c690/go.mod h1:O4p7UmsfoeLqtAUwrKbq0lXMxjY/MLQSpZSavvvvGig=
 github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
 github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=