diff --git a/address.go b/address.go
new file mode 100644
index 0000000..dadf5d3
--- /dev/null
+++ b/address.go
@@ -0,0 +1,34 @@
+package rabbit
+
+import (
+	"errors"
+	"fmt"
+	"net/url"
+)
+
+type Address struct {
+	Username string
+	Password string
+	Host     string
+	Port     uint16
+	Vhost    string
+}
+
+func (a *Address) makeAddr() (string, error) {
+	if a.Host == "" {
+		return "", errors.New("no host provided")
+	}
+
+	if a.Vhost == "" {
+		return "", errors.New("no vhost provided")
+	}
+
+	//"amqp://username:password@host:port/vhost"
+	return fmt.Sprintf("amqp://%v:%v@%v:%v/%v",
+		url.QueryEscape(a.Username),
+		url.QueryEscape(a.Password),
+		a.Host,
+		a.Port,
+		url.PathEscape(a.Vhost),
+	), nil
+}
diff --git a/consumer.go b/consumer.go
index b8d93d9..5c68bf6 100644
--- a/consumer.go
+++ b/consumer.go
@@ -4,18 +4,42 @@ import (
 	"context"
 	amqp "github.com/rabbitmq/amqp091-go"
 	"golang.org/x/time/rate"
-	"log"
+	"time"
 )
 
+type Consumer interface {
+	Start(ctx context.Context, chanLen uint) <-chan []byte
+}
+
+type consumeHandler struct {
+	client *Client
+}
+
+func (c *consumeHandler) Start(ctx context.Context, chanLen uint) <-chan []byte {
+	msgCh := make(chan []byte, chanLen)
+	go runConsumer(ctx, c.client, msgCh)
+
+	return msgCh
+}
+
 func runConsumer(ctx context.Context, client *Client, msgCh chan []byte) {
 	runCtx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
 	limiter := rate.NewLimiter(rate.Every(client.opts.consumerRateLimit), client.opts.consumerBurstSize)
 
-	deliveries, err := client.Consume()
+	reconnectSignal := make(chan struct{}, 1)
+	reconnectTrigger := func() {
+		select {
+		case reconnectSignal <- struct{}{}:
+		default:
+		}
+	}
+
+	// initial consume
+	deliveries, err := client.consume()
 	if err != nil {
-		log.Printf("Could not start consuming: %s\n", err)
+		client.logger.Printf("Could not start consuming: %s\n", err)
 		return
 	}
 
@@ -24,33 +48,66 @@ func runConsumer(ctx context.Context, client *Client, msgCh chan []byte) {
 
 	for {
 		select {
+
 		case <-runCtx.Done():
 			err = client.Close()
 			if err != nil {
-				log.Printf("Close failed: %s\n", err)
+				client.logger.Printf("Close failed: %s\n", err)
 			}
 			return
 
-		case amqErr := <-chClosedCh:
-			log.Printf("AMQP Channel closed due to: %s\n", amqErr)
+		case <-reconnectSignal:
+			select {
+			case <-runCtx.Done():
+				return
+			case <-time.After(client.opts.reconnectDelay):
+			}
 
-			deliveries, err = client.Consume()
+			deliveries, err = client.consume()
 			if err != nil {
-				log.Println("Error trying to consume, will try again")
+				client.logger.Printf("Consume reconnect failed, retry: %s\n", err)
+				reconnectTrigger()
 				continue
 			}
 
+			// re-create closing chan
 			chClosedCh = make(chan *amqp.Error, 1)
 			client.Channel.NotifyClose(chClosedCh)
+			client.logger.Println("Consume reconnect success")
 
-		case delivery := <-deliveries:
-			msgCh <- delivery.Body
-			log.Printf("Received message: %s\n", delivery.Body)
+		case amqErr := <-chClosedCh:
+			client.logger.Printf("AMQP Channel closed due to: %s Reconnecting...\n", amqErr)
+			reconnectTrigger()
 
-			if err = delivery.Ack(false); err != nil {
-				log.Printf("Error acknowledging message: %s\n", err)
+		case delivery, ok := <-deliveries:
+			if !ok {
+				client.logger.Println("Deliveries channel closed unexpectedly")
+				reconnectTrigger()
+				continue
+			}
+
+			if err = limiter.Wait(runCtx); err != nil {
+				client.logger.Printf("Wait limiter failed: %s\n", err)
+			}
+
+			select {
+			case <-runCtx.Done():
+				if err = delivery.Nack(false, true); err != nil {
+					client.logger.Printf("Error nacking message: %s\n", err)
+				}
+
+				err = client.Close()
+				if err != nil {
+					client.logger.Printf("Close failed: %s\n", err)
+				}
+				return
+
+			case msgCh <- delivery.Body:
+				client.logger.Printf("Received message: %s\n", delivery.Body)
+				if err = delivery.Ack(false); err != nil {
+					client.logger.Printf("Error acknowledging message: %s\n", err)
+				}
 			}
-			limiter.Wait(runCtx)
 		}
 	}
 }
diff --git a/handler.go b/handler.go
index 6109124..6a8dd3f 100644
--- a/handler.go
+++ b/handler.go
@@ -1,8 +1,10 @@
 package rabbit
 
 import (
-	"context"
+	"errors"
+	"fmt"
 	amqp "github.com/rabbitmq/amqp091-go"
+	"io"
 	"log"
 	"os"
 	"sync"
@@ -11,7 +13,7 @@ import (
 
 type Client struct {
 	mutex           *sync.Mutex
-	queueName       string
+	queueOptions    *QueueOpts
 	logger          *log.Logger
 	connection      *amqp.Connection
 	Channel         *amqp.Channel
@@ -21,52 +23,108 @@ type Client struct {
 	notifyConfirm   chan amqp.Confirmation
 	isReady         bool
 	opts            options
+	connected       chan struct{}
 }
 
 type options struct {
+	connectTimeout    time.Duration
 	reconnectDelay    time.Duration
 	reInitDelay       time.Duration
 	resendDelay       time.Duration
 	consumerRateLimit time.Duration
 	consumerBurstSize int
+	logger            *log.Logger
 }
 
-func NewClient(addr, queueName string, opts ...Option) (*Client, error) {
-	if addr == "" {
-		return nil, errNoAddr
+type QueueOpts struct {
+	QueueName  string
+	Durable    bool
+	AutoDelete bool
+	Exclusive  bool
+	NoWait     bool
+	Args       amqp.Table
+}
+
+func NewClient(address Address, queueOpts QueueOpts, opts ...Option) (*Client, error) {
+	l := log.New(os.Stdout, "", log.LstdFlags)
+
+	addr, err := address.makeAddr()
+	if err != nil {
+		return nil, errors.Join(errBadAddr, err)
 	}
 
-	if queueName == "" {
-		return nil, errNoQueue
+	if queueOpts.QueueName == "" {
+		l.Fatal(errNoQueue)
 	}
 
 	client := Client{
-		mutex:     &sync.Mutex{},
-		logger:    log.New(os.Stdout, "", log.LstdFlags),
-		queueName: queueName,
-		done:      make(chan bool),
+		mutex:        &sync.Mutex{},
+		queueOptions: &queueOpts,
+		done:         make(chan bool),
+		connected:    make(chan struct{}),
+		logger:       l,
 	}
 
 	o := options{
-		reconnectDelay:    5,
-		reInitDelay:       2,
-		resendDelay:       5,
+		connectTimeout:    15 * time.Second,
+		reconnectDelay:    5 * time.Second,
+		reInitDelay:       2 * time.Second,
+		resendDelay:       5 * time.Second,
 		consumerRateLimit: time.Millisecond * 500,
 		consumerBurstSize: 10,
+		logger:            log.New(io.Discard, "", 0),
 	}
 
 	for _, opt := range opts {
 		opt(&o)
 	}
 
+	client.logger = o.logger
+
+	if err = client.connectAndSignal(addr, o.connectTimeout); err != nil {
+		return nil, fmt.Errorf("failed to connect: %w", err)
+	}
+
 	go client.handleReconnect(addr)
 
 	return &client, nil
 }
 
-func StartConsumer(ctx context.Context, client *Client, chanLen int) chan []byte {
-	msgCh := make(chan []byte, chanLen)
-	go runConsumer(ctx, client, msgCh)
-
-	return msgCh
+func NewPublisher(client *Client) Publisher {
+	return &pubHandler{client: client}
+}
+
+func NewConsumer(client *Client) Consumer {
+	return &consumeHandler{client: client}
+}
+
+func (c *Client) connectAndSignal(addr string, timeout time.Duration) error {
+	type result struct {
+		conn *amqp.Connection
+		err  error
+	}
+	resCh := make(chan result, 1)
+
+	go func() {
+		conn, err := amqp.Dial(addr)
+		resCh <- result{conn, err}
+	}()
+
+	select {
+	case <-time.After(timeout):
+		return fmt.Errorf("connection timeout after %v", timeout)
+	case res := <-resCh:
+		if res.err != nil {
+			return res.err
+		}
+
+		c.changeConnection(res.conn)
+		if err := c.init(res.conn); err != nil {
+			res.conn.Close()
+			return fmt.Errorf("init failed: %w", err)
+		}
+
+		close(c.connected)
+		return nil
+	}
 }
diff --git a/messages.go b/messages.go
index 0f634d1..fd422ab 100644
--- a/messages.go
+++ b/messages.go
@@ -3,7 +3,7 @@ package rabbit
 import "errors"
 
 var (
-	errNoAddr        = errors.New("no address")
+	errBadAddr       = errors.New("bad address")
 	errNoQueue       = errors.New("no queue")
 	errNotConnected  = errors.New("not connected to a server")
 	errAlreadyClosed = errors.New("already closed: not connected to the server")
diff --git a/options.go b/options.go
index aaa7828..3e84f95 100644
--- a/options.go
+++ b/options.go
@@ -1,9 +1,20 @@
 package rabbit
 
-import "time"
+import (
+	"io"
+	"log"
+	"os"
+	"time"
+)
 
 type Option func(*options)
 
+func WithConnectTimeout(t time.Duration) Option {
+	return func(op *options) {
+		op.connectTimeout = t
+	}
+}
+
 func WithReconnectDelay(t time.Duration) Option {
 	return func(op *options) {
 		op.reconnectDelay = t
@@ -33,3 +44,17 @@ func WithConsumerBurstSize(t int) Option {
 		op.consumerBurstSize = t
 	}
 }
+
+func WithLogger(l *log.Logger) Option {
+	return func(op *options) { op.logger = l }
+}
+
+func WithLogging(enabled bool) Option {
+	return func(op *options) {
+		if enabled {
+			op.logger = log.New(os.Stdout, "", log.LstdFlags)
+		} else {
+			op.logger = log.New(io.Discard, "", 0)
+		}
+	}
+}
diff --git a/publisher.go b/publisher.go
new file mode 100644
index 0000000..c938eb5
--- /dev/null
+++ b/publisher.go
@@ -0,0 +1,65 @@
+package rabbit
+
+import (
+	"context"
+	"errors"
+	"time"
+)
+
+type Publisher interface {
+	Start(ctx context.Context, chanLen uint) chan<- []byte
+}
+
+type pubHandler struct {
+	client *Client
+}
+
+func (p *pubHandler) Start(ctx context.Context, chanLen uint) chan<- []byte {
+	ch := make(chan []byte, chanLen)
+	go func() {
+		for {
+			select {
+			case <-ctx.Done():
+				for msg := range ch {
+					if err := p.push(msg); err != nil {
+						p.client.logger.Printf("Error publishing message (shutdown): %s", err)
+					}
+				}
+				p.client.logger.Println("Publisher stopped")
+				return
+
+			case msg := <-ch:
+				if err := p.push(msg); err != nil {
+					p.client.logger.Printf("Error publishing message: %s", err)
+				}
+			}
+		}
+	}()
+	return ch
+}
+
+func (p *pubHandler) push(data []byte) error {
+	p.client.mutex.Lock()
+	if !p.client.isReady {
+		p.client.mutex.Unlock()
+		return errors.New("failed to push: not connected")
+	}
+	p.client.mutex.Unlock()
+	for {
+		err := p.client.unsafePush(data)
+		if err != nil {
+			p.client.logger.Println("Push failed. Retrying...")
+			select {
+			case <-p.client.done:
+				return errShutdown
+			case <-time.After(p.client.opts.resendDelay):
+			}
+			continue
+		}
+		confirm := <-p.client.notifyConfirm
+		if confirm.Ack {
+			p.client.logger.Printf("Push confirmed [%d]!", confirm.DeliveryTag)
+			return nil
+		}
+	}
+}
diff --git a/service.go b/service.go
index 889e8c7..c3f9ec8 100644
--- a/service.go
+++ b/service.go
@@ -2,17 +2,12 @@ package rabbit
 
 import (
 	"context"
-	"errors"
 	amqp "github.com/rabbitmq/amqp091-go"
 	"time"
 )
 
 func (c *Client) handleReconnect(addr string) {
 	for {
-		c.mutex.Lock()
-		c.isReady = false
-		c.mutex.Unlock()
-
 		c.logger.Println("Connecting to server...")
 
 		conn, err := c.connect(addr)
@@ -46,10 +41,6 @@ func (c *Client) connect(addr string) (*amqp.Connection, error) {
 
 func (c *Client) handleReInit(conn *amqp.Connection) bool {
 	for {
-		c.mutex.Lock()
-		c.isReady = false
-		c.mutex.Unlock()
-
 		if err := c.init(conn); err != nil {
 			c.logger.Printf("Failed to initialize connection: %s", err)
 
@@ -87,7 +78,16 @@ func (c *Client) init(conn *amqp.Connection) error {
 		return err
 	}
 
-	_, err = ch.QueueDeclare(c.queueName, false, false, false, false, nil)
+	//_, err = ch.QueueDeclare(c.queueName, false, false, false, false, nil)
+	_, err = ch.QueueDeclare(
+		c.queueOptions.QueueName,
+		c.queueOptions.Durable,
+		c.queueOptions.AutoDelete,
+		c.queueOptions.Exclusive,
+		c.queueOptions.NoWait,
+		c.queueOptions.Args,
+	)
+
 	if err != nil {
 		return err
 	}
@@ -115,33 +115,7 @@ func (c *Client) changeChannel(channel *amqp.Channel) {
 	c.Channel.NotifyPublish(c.notifyConfirm)
 }
 
-func (c *Client) Push(data []byte) error {
-	c.mutex.Lock()
-	if !c.isReady {
-		c.mutex.Unlock()
-		return errors.New("failed to push: not connected")
-	}
-	c.mutex.Unlock()
-	for {
-		err := c.UnsafePush(data)
-		if err != nil {
-			c.logger.Println("Push failed. Retrying...")
-			select {
-			case <-c.done:
-				return errShutdown
-			case <-time.After(c.opts.resendDelay):
-			}
-			continue
-		}
-		confirm := <-c.notifyConfirm
-		if confirm.Ack {
-			c.logger.Printf("Push confirmed [%d]!", confirm.DeliveryTag)
-			return nil
-		}
-	}
-}
-
-func (c *Client) UnsafePush(data []byte) error {
+func (c *Client) unsafePush(data []byte) error {
 	c.mutex.Lock()
 	if !c.isReady {
 		c.mutex.Unlock()
@@ -155,7 +129,7 @@ func (c *Client) UnsafePush(data []byte) error {
 	return c.Channel.PublishWithContext(
 		ctx,
 		"",
-		c.queueName,
+		c.queueOptions.QueueName,
 		false,
 		false,
 		amqp.Publishing{
@@ -165,7 +139,7 @@ func (c *Client) UnsafePush(data []byte) error {
 	)
 }
 
-func (c *Client) Consume() (<-chan amqp.Delivery, error) {
+func (c *Client) consume() (<-chan amqp.Delivery, error) {
 	c.mutex.Lock()
 	if !c.isReady {
 		c.mutex.Unlock()
@@ -182,7 +156,7 @@ func (c *Client) Consume() (<-chan amqp.Delivery, error) {
 	}
 
 	return c.Channel.Consume(
-		c.queueName,
+		c.queueOptions.QueueName,
 		"",
 		false,
 		false,