From 0440ccf4ffaae257dd8bfdbe279ca2408443880d Mon Sep 17 00:00:00 2001 From: 9seconds Date: Wed, 18 Feb 2026 10:33:46 +0100 Subject: [PATCH] Rewrite to WaitGroup.Go --- events/multi_observer.go | 81 ++++++++---------------- internal/cli/access.go | 14 ++-- ipblocklist/firehol.go | 11 ++-- network/circuit_breaker_internal_test.go | 17 ++--- network/network.go | 16 ++--- 5 files changed, 46 insertions(+), 93 deletions(-) diff --git a/events/multi_observer.go b/events/multi_observer.go index 2b529cd..f59d63e 100644 --- a/events/multi_observer.go +++ b/events/multi_observer.go @@ -12,14 +12,11 @@ type multiObserver struct { func (m multiObserver) EventStart(evt mtglib.EventStart) { wg := &sync.WaitGroup{} - wg.Add(len(m.observers)) for _, v := range m.observers { - go func(obs Observer) { - defer wg.Done() - - obs.EventStart(evt) - }(v) + wg.Go(func() { + v.EventStart(evt) + }) } wg.Wait() @@ -27,14 +24,11 @@ func (m multiObserver) EventStart(evt mtglib.EventStart) { func (m multiObserver) EventConnectedToDC(evt mtglib.EventConnectedToDC) { wg := &sync.WaitGroup{} - wg.Add(len(m.observers)) for _, v := range m.observers { - go func(obs Observer) { - defer wg.Done() - - obs.EventConnectedToDC(evt) - }(v) + wg.Go(func() { + v.EventConnectedToDC(evt) + }) } wg.Wait() @@ -42,14 +36,11 @@ func (m multiObserver) EventConnectedToDC(evt mtglib.EventConnectedToDC) { func (m multiObserver) EventDomainFronting(evt mtglib.EventDomainFronting) { wg := &sync.WaitGroup{} - wg.Add(len(m.observers)) for _, v := range m.observers { - go func(obs Observer) { - defer wg.Done() - - obs.EventDomainFronting(evt) - }(v) + wg.Go(func() { + v.EventDomainFronting(evt) + }) } wg.Wait() @@ -57,14 +48,11 @@ func (m multiObserver) EventDomainFronting(evt mtglib.EventDomainFronting) { func (m multiObserver) EventTraffic(evt mtglib.EventTraffic) { wg := &sync.WaitGroup{} - wg.Add(len(m.observers)) for _, v := range m.observers { - go func(obs Observer) { - defer wg.Done() - - obs.EventTraffic(evt) - }(v) + wg.Go(func() { + v.EventTraffic(evt) + }) } wg.Wait() @@ -72,14 +60,11 @@ func (m multiObserver) EventTraffic(evt mtglib.EventTraffic) { func (m multiObserver) EventFinish(evt mtglib.EventFinish) { wg := &sync.WaitGroup{} - wg.Add(len(m.observers)) for _, v := range m.observers { - go func(obs Observer) { - defer wg.Done() - - obs.EventFinish(evt) - }(v) + wg.Go(func() { + v.EventFinish(evt) + }) } wg.Wait() @@ -87,14 +72,11 @@ func (m multiObserver) EventFinish(evt mtglib.EventFinish) { func (m multiObserver) EventConcurrencyLimited(evt mtglib.EventConcurrencyLimited) { wg := &sync.WaitGroup{} - wg.Add(len(m.observers)) for _, v := range m.observers { - go func(obs Observer) { - defer wg.Done() - - obs.EventConcurrencyLimited(evt) - }(v) + wg.Go(func() { + v.EventConcurrencyLimited(evt) + }) } wg.Wait() @@ -102,14 +84,11 @@ func (m multiObserver) EventConcurrencyLimited(evt mtglib.EventConcurrencyLimite func (m multiObserver) EventIPBlocklisted(evt mtglib.EventIPBlocklisted) { wg := &sync.WaitGroup{} - wg.Add(len(m.observers)) for _, v := range m.observers { - go func(obs Observer) { - defer wg.Done() - - obs.EventIPBlocklisted(evt) - }(v) + wg.Go(func() { + v.EventIPBlocklisted(evt) + }) } wg.Wait() @@ -117,14 +96,11 @@ func (m multiObserver) EventIPBlocklisted(evt mtglib.EventIPBlocklisted) { func (m multiObserver) EventReplayAttack(evt mtglib.EventReplayAttack) { wg := &sync.WaitGroup{} - wg.Add(len(m.observers)) for _, v := range m.observers { - go func(obs Observer) { - defer wg.Done() - - obs.EventReplayAttack(evt) - }(v) + wg.Go(func() { + v.EventReplayAttack(evt) + }) } wg.Wait() @@ -132,14 +108,11 @@ func (m multiObserver) EventReplayAttack(evt mtglib.EventReplayAttack) { func (m multiObserver) EventIPListSize(evt mtglib.EventIPListSize) { wg := &sync.WaitGroup{} - wg.Add(len(m.observers)) for _, v := range m.observers { - go func(obs Observer) { - defer wg.Done() - - obs.EventIPListSize(evt) - }(v) + wg.Go(func() { + v.EventIPListSize(evt) + }) } wg.Wait() diff --git a/internal/cli/access.go b/internal/cli/access.go index 2e99c98..4657822 100644 --- a/internal/cli/access.go +++ b/internal/cli/access.go @@ -61,11 +61,8 @@ func (a *Access) Run(cli *CLI, version string) error { } wg := &sync.WaitGroup{} - wg.Add(2) - - go func() { - defer wg.Done() + wg.Go(func() { ip := a.PublicIPv4 if ip == nil { ip = a.getIP(ntw, "tcp4") @@ -76,11 +73,8 @@ func (a *Access) Run(cli *CLI, version string) error { } resp.IPv4 = a.makeURLs(conf, ip) - }() - - go func() { - defer wg.Done() - + }) + wg.Go(func() { ip := a.PublicIPv6 if ip == nil { ip = a.getIP(ntw, "tcp6") @@ -91,7 +85,7 @@ func (a *Access) Run(cli *CLI, version string) error { } resp.IPv6 = a.makeURLs(conf, ip) - }() + }) wg.Wait() diff --git a/ipblocklist/firehol.go b/ipblocklist/firehol.go index c2682d6..f219560 100644 --- a/ipblocklist/firehol.go +++ b/ipblocklist/firehol.go @@ -112,18 +112,15 @@ func (f *Firehol) update() { defer cancel() wg := &sync.WaitGroup{} - wg.Add(len(f.blocklists)) mutex := &sync.Mutex{} ranger := cidranger.NewPCTrieRanger() for _, v := range f.blocklists { - go func(file files.File) { - defer wg.Done() + wg.Go(func() { + logger := f.logger.BindStr("filename", v.String()) - logger := f.logger.BindStr("filename", file.String()) - - fileContent, err := file.Open(ctx) + fileContent, err := v.Open(ctx) if err != nil { logger.WarningError("update has failed", err) @@ -135,7 +132,7 @@ func (f *Firehol) update() { if err := f.updateFromFile(mutex, ranger, bufio.NewScanner(fileContent)); err != nil { logger.WarningError("update has failed", err) } - }(v) + }) } wg.Wait() diff --git a/network/circuit_breaker_internal_test.go b/network/circuit_breaker_internal_test.go index fc7f671..a665e21 100644 --- a/network/circuit_breaker_internal_test.go +++ b/network/circuit_breaker_internal_test.go @@ -52,17 +52,9 @@ func (suite *CircuitBreakerTestSuite) TestMultipleRunsOk() { Return(suite.connMock, nil) wg := &sync.WaitGroup{} - wg.Add(5) - - go func() { - wg.Wait() - suite.ctxCancel() - }() for range 5 { - go func() { - defer wg.Done() - + wg.Go(func() { conn, err := suite.d.DialContext(suite.ctx, "tcp", "127.0.0.1") suite.mutex.Lock() @@ -70,9 +62,14 @@ func (suite *CircuitBreakerTestSuite) TestMultipleRunsOk() { suite.NoError(err) suite.Equal("127.0.0.1:3128", conn.RemoteAddr().String()) - }() + }) } + go func() { + wg.Wait() + suite.ctxCancel() + }() + suite.Eventually(func() bool { _, ok := <-suite.ctx.Done() diff --git a/network/network.go b/network/network.go index 5b536b4..7a7b332 100644 --- a/network/network.go +++ b/network/network.go @@ -81,32 +81,24 @@ func (n *network) dnsResolve(protocol, address string) ([]string, error) { switch protocol { case "tcp", "tcp4": - wg.Add(1) - - go func() { - defer wg.Done() - + wg.Go(func() { resolved := n.dns.LookupA(address) mutex.Lock() ips = append(ips, resolved...) mutex.Unlock() - }() + }) } switch protocol { case "tcp", "tcp6": - wg.Add(1) - - go func() { - defer wg.Done() - + wg.Go(func() { resolved := n.dns.LookupAAAA(address) mutex.Lock() ips = append(ips, resolved...) mutex.Unlock() - }() + }) } wg.Wait()