class RMQSource extends MultipleIdsMessageAcknowledgingSourceBase<OUT, String, Long> implements ResultTypeQueryable<OUT> {

	@Override
	public void open(Configuration config) throws Exception {
		super.open(config);
		ConnectionFactory factory = setupConnectionFactory();
		try {
			connection = factory.newConnection();
			channel = connection.createChannel();
			if (channel == null) {
				throw new RuntimeException("None of RabbitMQ channels are available");
			}
			setupQueue();
			consumer = new QueueingConsumer(channel);

			RuntimeContext runtimeContext = getRuntimeContext();
			if (runtimeContext instanceof StreamingRuntimeContext
					&& ((StreamingRuntimeContext) runtimeContext).isCheckpointingEnabled()) {
				autoAck = false;
				// enables transaction mode
				channel.txSelect();
			} else {
				autoAck = true;
			}

			LOG.debug("Starting RabbitMQ source with autoAck status: " + autoAck);
			channel.basicConsume(queueName, autoAck, consumer);

		} catch (IOException e) {
			throw new RuntimeException("Cannot create RMQ connection with " + queueName + " at "
					+ rmqConnectionConfig.getHost(), e);
		}
		running = true;
	}

	public  RMQSource(RMQConnectionConfig rmqConnectionConfig, String queueName,
					DeserializationSchema<OUT> deserializationSchema);
	public  RMQSource(RMQConnectionConfig rmqConnectionConfig,
					String queueName, boolean usesCorrelationId,DeserializationSchema<OUT> deserializationSchema);

}