remove set addr from publisher config

andyollylarkin · Aug 25, 2023 · 0d0181f · 0d0181f
1 parent dda2416
commit 0d0181f
Show file tree

Hide file tree

Showing 5 changed files with 10 additions and 42 deletions.
diff --git a/examples/client/client.go b/examples/client/client.go
@@ -16,7 +16,6 @@ func main() {
 	pConn := connection.NewTCPConnection(net.Dialer{}, time.Second*30)
 
 	p, err := watermillnet.NewPublisher(watermillnet.PublisherConfig{
-		RemoteAddr:  &net.TCPAddr{IP: []byte{127, 0, 0, 1}, Port: 9090},
 		Marshaler:   pkg.MessagePackMarshaler{},
 		Unmarshaler: pkg.MessagePackUnmarshaler{},
 	}, false)
@@ -26,7 +25,7 @@ func main() {
 		log.Fatal(err)
 	}
 
-	err = p.Connect()
+	err = p.Connect(&net.TCPAddr{IP: []byte{127, 0, 0, 1}, Port: 9090})
 
 	if err != nil {
 		log.Fatal(err)

diff --git a/examples/retryable_client/client.go b/examples/retryable_client/client.go
@@ -24,7 +24,6 @@ func main() {
 		watermill.NewStdLogger(true, true), addr, connection.DefaultErrorFilter, time.Second*4)
 
 	p, err := watermillnet.NewPublisher(watermillnet.PublisherConfig{
-		RemoteAddr:  addr,
 		Marshaler:   pkg.MessagePackMarshaler{},
 		Unmarshaler: pkg.MessagePackUnmarshaler{},
 		Logger:      watermill.NewStdLogger(true, true),
@@ -35,7 +34,7 @@ func main() {
 
 	p.SetConnection(wpConn)
 
-	err = p.Connect()
+	err = p.Connect(addr)
 
 	if err != nil {
 		log.Fatal(err)

diff --git a/publisher.go b/publisher.go
@@ -17,7 +17,6 @@ import (
 )
 
 type PublisherConfig struct {
-	RemoteAddr  net.Addr
 	Marshaler   Marshaler
 	Unmarshaler Unmarshaler
 	Logger      watermill.LoggerAdapter
@@ -27,7 +26,6 @@ type Publisher struct {
 	conn        Connection
 	marshaler   Marshaler
 	unmarshaler Unmarshaler
-	addr        net.Addr
 	logger      watermill.LoggerAdapter
 	closed      bool
 	mu          sync.Mutex
@@ -42,7 +40,6 @@ func NewPublisher(config PublisherConfig, waitAck bool) (*Publisher, error) {
 	}
 
 	p := new(Publisher)
-	p.addr = config.RemoteAddr
 	p.marshaler = config.Marshaler
 	p.unmarshaler = config.Unmarshaler
 	p.logger = config.Logger
@@ -52,10 +49,6 @@ func NewPublisher(config PublisherConfig, waitAck bool) (*Publisher, error) {
 }
 
 func validatePublisherConfig(c PublisherConfig) error {
-	if c.RemoteAddr == nil {
-		return &InvalidConfigError{InvalidField: "Addr", InvalidReason: "cant be nil"}
-	}
-
 	if c.Marshaler == nil {
 		return &InvalidConfigError{InvalidField: "Marshaler", InvalidReason: "cant be nil"}
 	}
@@ -91,7 +84,7 @@ func (p *Publisher) GetConnection() (Connection, error) {
 }
 
 // Connect to remote side.
-func (p *Publisher) Connect() error {
+func (p *Publisher) Connect(addr net.Addr) error {
 	if p.closed {
 		return ErrPublisherClosed
 	}
@@ -100,7 +93,7 @@ func (p *Publisher) Connect() error {
 		return ErrConnectionNotSet
 	}
 
-	return p.conn.Connect(p.addr)
+	return p.conn.Connect(addr)
 }
 
 // Publish publishes provided messages to given topic.

diff --git a/publisher_test.go b/publisher_test.go
@@ -41,7 +41,6 @@ func TestPublishMessageRemoteSideReceiveOK(t *testing.T) {
 	pipeConn := NewPipeConnection()
 	uuid := watermill.NewUUID()
 	config := watermillnet.PublisherConfig{
-		RemoteAddr:  pipeAddr{},
 		Marshaler:   pkg.MessagePackMarshaler{},
 		Unmarshaler: pkg.MessagePackUnmarshaler{},
 	}
@@ -65,7 +64,6 @@ func TestPublishMessageRemoteSideReceiveNackResponse(t *testing.T) {
 	pipeConn := NewPipeConnection()
 	uuid := watermill.NewUUID()
 	config := watermillnet.PublisherConfig{
-		RemoteAddr:  pipeAddr{},
 		Marshaler:   pkg.MessagePackMarshaler{},
 		Unmarshaler: pkg.MessagePackUnmarshaler{},
 	}
@@ -91,7 +89,6 @@ func TestPublishMessageOnClosedPublisher(t *testing.T) {
 	pipeConn := NewPipeConnection()
 	uuid := watermill.NewUUID()
 	config := watermillnet.PublisherConfig{
-		RemoteAddr:  pipeAddr{},
 		Marshaler:   pkg.MessagePackMarshaler{},
 		Unmarshaler: pkg.MessagePackUnmarshaler{},
 	}
@@ -121,19 +118,9 @@ func TestPublisherCreateError(t *testing.T) {
 		config   watermillnet.PublisherConfig
 		expected string
 	}{
-		{
-			name: "Err Addr nil",
-			config: watermillnet.PublisherConfig{
-				RemoteAddr:  nil,
-				Marshaler:   pkg.MessagePackMarshaler{},
-				Unmarshaler: pkg.MessagePackUnmarshaler{},
-			},
-			expected: "invalid field: Addr. reason: cant be nil",
-		},
 		{
 			name: "Err Marshaler nil",
 			config: watermillnet.PublisherConfig{
-				RemoteAddr:  pipeAddr{},
 				Marshaler:   nil,
 				Unmarshaler: pkg.MessagePackUnmarshaler{},
 			},
@@ -142,7 +129,6 @@ func TestPublisherCreateError(t *testing.T) {
 		{
 			name: "Err Unmarshaler nil",
 			config: watermillnet.PublisherConfig{
-				RemoteAddr:  pipeAddr{},
 				Marshaler:   pkg.MessagePackMarshaler{},
 				Unmarshaler: nil,
 			},
@@ -168,7 +154,6 @@ func TestPublisherCreateOK(t *testing.T) {
 		{
 			name: "Publisher no error",
 			config: watermillnet.PublisherConfig{
-				RemoteAddr:  pipeAddr{},
 				Marshaler:   pkg.MessagePackMarshaler{},
 				Unmarshaler: pkg.MessagePackUnmarshaler{},
 			},
@@ -186,7 +171,6 @@ func TestPublisherCreateOK(t *testing.T) {
 func TestPublishToClosedPublisher(t *testing.T) {
 	conn := NewPipeConnection()
 	config := watermillnet.PublisherConfig{
-		RemoteAddr:  pipeAddr{},
 		Marshaler:   pkg.MessagePackMarshaler{},
 		Unmarshaler: pkg.MessagePackUnmarshaler{},
 	}
@@ -204,7 +188,6 @@ func TestPublishMultiMessage(t *testing.T) {
 	pipeConn := NewPipeConnection()
 	uuid := watermill.NewUUID()
 	config := watermillnet.PublisherConfig{
-		RemoteAddr:  pipeAddr{},
 		Marshaler:   pkg.MessagePackMarshaler{},
 		Unmarshaler: pkg.MessagePackUnmarshaler{},
 	}
@@ -232,7 +215,6 @@ func TestPublishMultiMessage(t *testing.T) {
 func TestConnectOnClosedPublisher(t *testing.T) {
 	conn := NewPipeConnection()
 	c := watermillnet.PublisherConfig{
-		RemoteAddr:  pipeAddr{},
 		Marshaler:   pkg.MessagePackMarshaler{},
 		Unmarshaler: pkg.MessagePackUnmarshaler{},
 	}
@@ -242,29 +224,27 @@ func TestConnectOnClosedPublisher(t *testing.T) {
 	assert.NoError(t, err)
 	err = p.Close()
 	assert.NoError(t, err)
-	err = p.Connect()
+	err = p.Connect(pipeAddr{})
 	assert.ErrorIs(t, err, watermillnet.ErrPublisherClosed)
 }
 
 func TestConnectOK(t *testing.T) {
 	conn := NewPipeConnection()
 	c := watermillnet.PublisherConfig{
-		RemoteAddr:  pipeAddr{},
 		Marshaler:   pkg.MessagePackMarshaler{},
 		Unmarshaler: pkg.MessagePackUnmarshaler{},
 	}
 
 	p, err := watermillnet.NewPublisher(c, false)
 	p.SetConnection(conn)
 	assert.NoError(t, err)
-	err = p.Connect()
+	err = p.Connect(pipeAddr{})
 	assert.NoError(t, err)
 }
 
 func TestPublisherConnectionNotSetError(t *testing.T) {
 	retPub := func(t *testing.T) *watermillnet.Publisher {
 		p, err := watermillnet.NewPublisher(watermillnet.PublisherConfig{
-			RemoteAddr:  pipeAddr{},
 			Marshaler:   pkg.MessagePackMarshaler{},
 			Unmarshaler: pkg.MessagePackUnmarshaler{},
 		}, false)
@@ -298,7 +278,7 @@ func TestPublisherConnectionNotSetError(t *testing.T) {
 		}, expect: watermillnet.ErrConnectionNotSet},
 		{name: "Connect", exec: func(t *testing.T) error {
 			p := retPub(t)
-			err := p.Connect()
+			err := p.Connect(pipeAddr{})
 
 			return err
 		}, expect: watermillnet.ErrConnectionNotSet},

diff --git a/subscriber_test.go b/subscriber_test.go
@@ -108,13 +108,12 @@ func TestPublisherConnectError(t *testing.T) {
 func TestCloseWaitAllConns(t *testing.T) {
 	pipeConn := NewPipeConnection()
 	p, err := watermillnet.NewPublisher(watermillnet.PublisherConfig{
-		RemoteAddr:  pipeAddr{},
 		Marshaler:   pkg.MessagePackMarshaler{},
 		Unmarshaler: pkg.MessagePackUnmarshaler{},
 	}, false)
 	p.SetConnection(pipeConn)
 	require.NoError(t, err)
-	err = p.Connect()
+	err = p.Connect(pipeAddr{})
 	require.NoError(t, err)
 
 	s, err := watermillnet.NewSubscriber(watermillnet.SubscriberConfig{
@@ -169,13 +168,12 @@ func TestCloseWaitAllConns(t *testing.T) {
 func TestCloseSocketConn(t *testing.T) {
 	pipeConn := NewPipeConnection()
 	p, err := watermillnet.NewPublisher(watermillnet.PublisherConfig{
-		RemoteAddr:  pipeAddr{},
 		Marshaler:   pkg.MessagePackMarshaler{},
 		Unmarshaler: pkg.MessagePackUnmarshaler{},
 	}, false)
 	p.SetConnection(pipeConn)
 	require.NoError(t, err)
-	err = p.Connect()
+	err = p.Connect(pipeAddr{})
 	require.NoError(t, err)
 
 	s, err := watermillnet.NewSubscriber(watermillnet.SubscriberConfig{
@@ -200,13 +198,12 @@ func TestCloseSocketConn(t *testing.T) {
 func TestSubscriberCancelByContext(t *testing.T) {
 	pipeConn := NewPipeConnection()
 	p, err := watermillnet.NewPublisher(watermillnet.PublisherConfig{
-		RemoteAddr:  pipeAddr{},
 		Marshaler:   pkg.MessagePackMarshaler{},
 		Unmarshaler: pkg.MessagePackUnmarshaler{},
 	}, false)
 	p.SetConnection(pipeConn)
 	require.NoError(t, err)
-	err = p.Connect()
+	err = p.Connect(pipeAddr{})
 	require.NoError(t, err)
 
 	s, err := watermillnet.NewSubscriber(watermillnet.SubscriberConfig{