return err
	}
	target.producer = producer

	err = target.producer.Ping()
	if err != nil {
		// To treat "connection refused" errors as errNotConnected.
		if !(xnet.IsConnRefusedErr(err) || xnet.IsConnResetErr(err)) {
			target.loggerOnce(context.Background(), err, target.ID().String())
		}
		target.producer.Stop()
		return err
	}

	yes, err := target.isActive()
	if err != nil {
		return err
	}

}

func (target *RedisTarget) initRedis() error {
	conn := target.pool.Get()
	defer conn.Close()

	_, pingErr := conn.Do("PING")
	if pingErr != nil {
		if !(xnet.IsConnRefusedErr(pingErr) || xnet.IsConnResetErr(pingErr)) {
			target.loggerOnce(context.Background(), pingErr, target.ID().String())
		}
		return pingErr
	}

	if err := target.args.validateFormat(conn); err != nil {

		// Set the maximum connections limit
		target.db.SetMaxOpenConns(args.MaxOpenConnections)
	}

	err = target.db.Ping()
	if err != nil {
		if !(xnet.IsConnRefusedErr(err) || xnet.IsConnResetErr(err)) {
			target.loggerOnce(context.Background(), err, target.ID().String())
		}
	} else {
		if err = target.executeStmts(); err != nil {
			target.loggerOnce(context.Background(), err, target.ID().String())

	return target.initOnce.Do(target.initAMQP)
}

func (target *AMQPTarget) initAMQP() error {
	conn, err := amqp091.Dial(target.args.URL.String())
	if err != nil {
		if xnet.IsConnRefusedErr(err) || xnet.IsConnResetErr(err) {
			target.loggerOnce(context.Background(), err, target.ID().String())
		}
		return err
	}
	target.conn = conn

	return nil
}

// NewAMQPTarget - creates new AMQP target.

		// Set the maximum connections limit
		target.db.SetMaxOpenConns(args.MaxOpenConnections)
	}

	err = target.db.Ping()
	if err != nil {
		if !(xnet.IsConnRefusedErr(err) || xnet.IsConnResetErr(err)) {
			target.loggerOnce(context.Background(), err, target.ID().String())
		}
	} else {
		if err = target.executeStmts(); err != nil {
			target.loggerOnce(context.Background(), err, target.ID().String())