updated to pipeline package v0.2.2

Author: caffix

enum/dns.go

@@ -12,11 +12,11 @@ import (
 	"sync"
 	"time"
 
-	"github.com/owasp-amass/amass/v3/requests"
 	"github.com/caffix/pipeline"
 	"github.com/caffix/queue"
-	"github.com/owasp-amass/resolve"
 	"github.com/miekg/dns"
+	"github.com/owasp-amass/amass/v3/requests"
+	"github.com/owasp-amass/resolve"
 )
 
 const (
@@ -239,7 +239,6 @@ func (dt *dnsTask) addReqWithIncrement(key string, entry *req) bool {
 
 	if added {
 		<-dt.release
-		_ = dt.params.Pipeline().IncDataItemCount()
 	}
 	return added
 }
@@ -259,7 +258,7 @@ func (dt *dnsTask) delReq(key string) *req {
 func (dt *dnsTask) delReqWithDecrement(key string) {
 	if req := dt.delReq(key); req != nil {
 		dt.release <- struct{}{}
-		_ = dt.params.Pipeline().DecDataItemCount()
+
 		if !req.Sent && (req.InScope || req.HasRecords) {
 			dt.nextStage(req.Ctx, req.Data)
 		}
@@ -417,8 +416,6 @@ func (dt *dnsTask) subdomainQueries(ctx context.Context, req *requests.DNSReques
 }
 
 func (dt *dnsTask) queryNS(ctx context.Context, name, domain string, ch chan []requests.DNSAnswer, tp pipeline.TaskParams) {
-	tp.Pipeline().IncDataItemCount()
-	defer tp.Pipeline().DecDataItemCount()
 	// Obtain the DNS answers for the NS records related to the domain
 	if resp, err := dt.enum.dnsQuery(ctx, name, dns.TypeNS, dt.enum.Sys.TrustedResolvers(), maxDNSQueryAttempts); err == nil {
 		if ans := resolve.ExtractAnswers(resp); len(ans) > 0 {
@@ -445,8 +442,6 @@ func (dt *dnsTask) queryNS(ctx context.Context, name, domain string, ch chan []r
 }
 
 func (dt *dnsTask) queryMX(ctx context.Context, name string, ch chan []requests.DNSAnswer, tp pipeline.TaskParams) {
-	tp.Pipeline().IncDataItemCount()
-	defer tp.Pipeline().DecDataItemCount()
 	// Obtain the DNS answers for the MX records related to the domain
 	if resp, err := dt.enum.dnsQuery(ctx, name, dns.TypeMX, dt.enum.Sys.TrustedResolvers(), maxDNSQueryAttempts); err == nil {
 		if ans := resolve.ExtractAnswers(resp); len(ans) > 0 {
@@ -460,8 +455,6 @@ func (dt *dnsTask) queryMX(ctx context.Context, name string, ch chan []requests.
 }
 
 func (dt *dnsTask) querySOA(ctx context.Context, name string, ch chan []requests.DNSAnswer, tp pipeline.TaskParams) {
-	tp.Pipeline().IncDataItemCount()
-	defer tp.Pipeline().DecDataItemCount()
 	// Obtain the DNS answers for the SOA records related to the domain
 	if resp, err := dt.enum.dnsQuery(ctx, name, dns.TypeSOA, dt.enum.Sys.TrustedResolvers(), maxDNSQueryAttempts); err == nil {
 		if ans := resolve.ExtractAnswers(resp); len(ans) > 0 {
@@ -481,8 +474,6 @@ func (dt *dnsTask) querySOA(ctx context.Context, name string, ch chan []requests
 }
 
 func (dt *dnsTask) querySPF(ctx context.Context, name string, ch chan []requests.DNSAnswer, tp pipeline.TaskParams) {
-	tp.Pipeline().IncDataItemCount()
-	defer tp.Pipeline().DecDataItemCount()
 	// Obtain the DNS answers for the SPF records related to the domain
 	if resp, err := dt.enum.dnsQuery(ctx, name, dns.TypeSPF, dt.enum.Sys.TrustedResolvers(), maxDNSQueryAttempts); err == nil {
 		if ans := resolve.ExtractAnswers(resp); len(ans) > 0 {

go.mod

@@ -5,7 +5,7 @@ go 1.19
 require (
 	github.com/PuerkitoBio/goquery v1.8.1
 	github.com/caffix/netmap v0.1.0
-	github.com/caffix/pipeline v0.2.1
+	github.com/caffix/pipeline v0.2.2
 	github.com/caffix/queue v0.1.4
 	github.com/caffix/service v0.3.0
 	github.com/caffix/stringset v0.1.1

go.sum

@@ -96,8 +96,8 @@ github.com/boltdb/bolt v1.3.1 h1:JQmyP4ZBrce+ZQu0dY660FMfatumYDLun9hBCUVIkF4=
 github.com/boltdb/bolt v1.3.1/go.mod h1:clJnj/oiGkjum5o1McbSZDSLxVThjynRyGBgiAx27Ps=
 github.com/caffix/netmap v0.1.0 h1:T+9FRmJqSVjfHoyTRGnLCQ0ZM/bc3qc0vlcUiIgaZAk=
 github.com/caffix/netmap v0.1.0/go.mod h1:O2MKqKckvo9kz0AfWuMNOg2poyW5eph/iINYLKI7EJ4=
-github.com/caffix/pipeline v0.2.1 h1:sfkBebseEmIh39+wHiaWLwCirx4sfTofUuT7Do0tbM8=
-github.com/caffix/pipeline v0.2.1/go.mod h1:0q0Dx1s1pIi7peIwz3aTIzQRmTEtH1n45NJyrTkiIHs=
+github.com/caffix/pipeline v0.2.2 h1:d1l7CiBe7jFDc4ksvnNgHVb3XlVlCRFXFo3I6guksuQ=
+github.com/caffix/pipeline v0.2.2/go.mod h1:NlUnsifT5h0B9C51DPdp99KjlZjREnE0Iev7R8gEn/s=
 github.com/caffix/queue v0.1.4 h1:sQbFzwGaPM1tRnQHWCgHOwj7hLuhDQ3BhY1/1TFbBiE=
 github.com/caffix/queue v0.1.4/go.mod h1:l8Eg7UTUHTRlc5aQ37mRVjzLN6eC7hgwimN0pA4UHe8=
 github.com/caffix/service v0.3.0 h1:Sb0GVFaYnn7mJCWyfcGr4AumdoHKT9+7gn6A96U88eY=