refactored

ignavan39 · Sep 20, 2023 · 7ea0d6f · 7ea0d6f
1 parent a6fa6ef
commit 7ea0d6f
Show file tree

Hide file tree

Showing 3 changed files with 52 additions and 40 deletions.
diff --git a/example/main.go b/example/main.go
@@ -25,7 +25,7 @@ func main() {
 
 	go func() {
 		for i := 0; i <= 20; i++ {
-			wq.WriteChannel <- piper.Message{
+			wq.Write() <- piper.Message{
 				Payload: i,
 				UID:     "uid",
 			}
@@ -37,9 +37,9 @@ func main() {
 	if err != nil {
 		panic(err)
 	}
-	for message := range c.MessagesChannel {
+	for message := range c.Read() {
 		fmt.Println(message)
-		c.Report.Report <- piper.Report{
+		c.Report() <- piper.Report{
 			Done: &piper.DoneReport{
 				Status: 1,
 			},

diff --git a/read_queue.go b/read_queue.go
@@ -11,19 +11,19 @@ import (
 
 type ReadQueueReport struct {
 	channel      *amqp.Channel
-	Report       chan Report
+	report       chan Report
 	exchangeName string
 	routingKey   string
 }
 
 type ReadQueue struct {
-	conn            *amqp.Connection
-	channel         *amqp.Channel
-	queue           string
-	done            chan error
-	routines        int
-	MessagesChannel chan Message
-	Report          *ReadQueueReport
+	conn     *amqp.Connection
+	channel  *amqp.Channel
+	queue    string
+	done     chan error
+	routines int
+	read     chan Message
+	report   *ReadQueueReport
 }
 
 func NewReadQueue(
@@ -68,12 +68,12 @@ func NewReadQueue(
 	channel.Qos(routines, 0, false)
 
 	return &ReadQueue{
-		channel:         channel,
-		conn:            conn,
-		queue:           queue,
-		done:            make(chan error),
-		routines:        routines,
-		MessagesChannel: make(chan Message),
+		channel:  channel,
+		conn:     conn,
+		queue:    queue,
+		done:     make(chan error),
+		routines: routines,
+		read:     make(chan Message),
 	}, nil
 }
 
@@ -96,17 +96,25 @@ func (rq *ReadQueue) WithReport(reportExchangeName string, reportRoutingKey stri
 		return nil, fmt.Errorf("[Rq][%s][createReportChannel]: %s", rq.queue, err)
 	}
 	report := make(chan Report)
-	rq.Report = &ReadQueueReport{
+	rq.report = &ReadQueueReport{
 		exchangeName: reportExchangeName,
 		channel:      reportChannel,
-		Report:       report,
+		report:       report,
 		routingKey:   reportRoutingKey,
 	}
 	return rq, nil
 }
 
+func (rq *ReadQueue) Report() chan Report {
+	return rq.report.report
+}
+
+func (rq *ReadQueue) Read() <-chan Message {
+	return rq.read
+}
+
 func (rq *ReadQueue) Run() error {
-	defer close(rq.MessagesChannel)
+	defer close(rq.read)
 	var wg sync.WaitGroup
 	deliveries, err := rq.channel.Consume(
 		rq.queue,
@@ -148,28 +156,28 @@ func (rq *ReadQueue) Run() error {
 							continue
 						}
 						fmt.Println(idx)
-						rq.MessagesChannel <- *message
+						rq.read <- *message
 					}
 				}
 			}(i)
 		}
 	}()
 
-	if rq.Report != nil {
+	if rq.report != nil {
 		wg.Add(1)
 		go func() {
-			defer close(rq.Report.Report)
+			defer close(rq.report.report)
 			defer wg.Done()
-			for report := range rq.Report.Report {
+			for report := range rq.report.report {
 				buffer, err := json.Marshal(report)
 				if err != nil {
 					fmt.Printf("[Rq][%s][Run][report] - failed marshal: %s", rq.queue, err)
 					continue
 				}
 
-				rq.Report.channel.Publish(
-					rq.Report.exchangeName,
-					rq.Report.routingKey,
+				rq.report.channel.Publish(
+					rq.report.exchangeName,
+					rq.report.routingKey,
 					false,
 					false,
 					amqp.Publishing{

diff --git a/write_queue.go b/write_queue.go
@@ -9,12 +9,12 @@ import (
 )
 
 type WriteQueue struct {
-	WriteChannel chan Message
-	conn         *amqp.Connection
-	channel      *amqp.Channel
-	exchange     string
-	routingKey   string
-	done         chan error
+	write      chan Message
+	conn       *amqp.Connection
+	channel    *amqp.Channel
+	exchange   string
+	routingKey string
+	done       chan error
 }
 
 func NewWriteQueue(
@@ -41,17 +41,21 @@ func NewWriteQueue(
 	}
 
 	wq := &WriteQueue{
-		WriteChannel: make(chan Message),
-		conn:         conn,
-		channel:      channel,
-		exchange:     exchange,
-		routingKey:   routingKey,
+		write:      make(chan Message),
+		conn:       conn,
+		channel:    channel,
+		exchange:   exchange,
+		routingKey: routingKey,
 	}
 	return wq, nil
 }
 
+func (wq *WriteQueue) Write() chan Message {
+	return wq.write
+}
+
 func (wq *WriteQueue) Run() {
-	defer close(wq.WriteChannel)
+	defer close(wq.write)
 	var wg sync.WaitGroup
 
 	wg.Add(1)
@@ -60,7 +64,7 @@ func (wq *WriteQueue) Run() {
 		defer wg.Done()
 		for {
 			select {
-			case payload, ok := <-wq.WriteChannel:
+			case payload, ok := <-wq.write:
 				fmt.Println(payload)
 				if !ok {
 					fmt.Printf("[Wq][%s-%s][Run][channel closed]\n", wq.exchange, wq.routingKey)