code: @Override
  	public void connect() {
  		try {
  
  			// Create client connection
  			disconnect();
  			String uri = url;
  			if (uri.indexOf(':') == -1) {
  				uri = uri + ":5672";
  			}
  			if (url.indexOf("://") == -1) {
  				if (sslContextFactory != null) {
  					uri = "amqps://" + uri;
  				} else {
  					uri = "amqp://" + uri;
  				}
  			}
  			factory.setHeartbeatExecutor(scheduler);
  			factory.setSharedExecutor(executor);
  			factory.setSslContextFactory(sslContextFactory);
  
  			factory.setAutomaticRecoveryEnabled(false);
  			factory.setTopologyRecoveryEnabled(false);
  
  			// NioParams params = new NioParams();
  			// params.setNioExecutor(executor);
  			// factory.setNioParams(params);
  
  			factory.setUri(uri);
  			if (username != null) {
  				factory.setUsername(username);
  			}
  			if (password != null) {
  				factory.setPassword(password);
  			}
  			started.set(true);
  			client = factory.newConnection();
  			channel = client.createChannel();
  
  			logger.info("AMQP pub-sub connection estabilished.");
  			connected();
  		} catch (Exception cause) {
  			String msg = cause.getMessage();
  			if (msg == null || msg.isEmpty()) {
  				msg = "Unable to connect to AMQP server!";
  			} else if (!msg.endsWith("!") && !msg.endsWith(".")) {
  				msg += "!";
  			}
  			logger.warn(msg);
  			reconnect();
  		}
  	}
docs: --- CONNECT ---