// Path: common/java/common/AMQPCommon.java
// public class AMQPCommon {
// 
// 	public static Channel connect() throws Exception {
// 		ConnectionFactory factory = new ConnectionFactory();
// 		factory.setHost("127.0.0.1");
// 		factory.setPort(32768);
// 		Connection conn = factory.newConnection();
// 		return conn.createChannel();
// 	}
// 	
// 	public static void close(Channel channel) throws Exception {
// 		channel.close();
// 		channel.getConnection().close();
// 	}
// 
// }


import com.rabbitmq.client.Channel;
import common.AMQPCommon;
package producerflow;


public class AMQPFlowMonitor {
	
	public void run() throws Exception {
		Channel channel = AMQPCommon.connect();
		long threshold = 10;
		boolean controlFlow = false;
		while (true) {
			long queueDepth = channel.messageCount("trade.eq.q");
			if (queueDepth > threshold && !controlFlow) {
				controlFlow = enableControlFlow(channel);
			} else if (queueDepth <= (threshold/2) && controlFlow) {
				controlFlow = disableControlFlow(channel);
			}
			Thread.sleep(3000);
		}
	}
	
	private boolean enableControlFlow(Channel channel) throws Exception {
		System.out.println("Enabling producer control flow...");
		byte[] msg = String.valueOf(true).getBytes();
		channel.basicPublish("flow.fx", "", null, msg);
		return true;
	}
	
// Path: common/java/common/AMQPCommon.java
// public class AMQPCommon {
// 
// 	public static Channel connect() throws Exception {
// 		ConnectionFactory factory = new ConnectionFactory();
// 		factory.setHost("127.0.0.1");
// 		factory.setPort(32768);
// 		Connection conn = factory.newConnection();
// 		return conn.createChannel();
// 	}
// 	
// 	public static void close(Channel channel) throws Exception {
// 		channel.close();
// 		channel.getConnection().close();
// 	}
// 
// }


// Path: producercontrolflow/java/producerflow/AMQPFlowMonitor.java
import com.rabbitmq.client.Channel;
import common.AMQPCommon;

package producerflow;


public class AMQPFlowMonitor {
	
	public void run() throws Exception {
		Channel channel = AMQPCommon.connect();
		long threshold = 10;
		boolean controlFlow = false;
		while (true) {
			long queueDepth = channel.messageCount("trade.eq.q");
			if (queueDepth > threshold && !controlFlow) {
				controlFlow = enableControlFlow(channel);
			} else if (queueDepth <= (threshold/2) && controlFlow) {
				controlFlow = disableControlFlow(channel);
			}
			Thread.sleep(3000);
		}
	}
	
	private boolean enableControlFlow(Channel channel) throws Exception {
		System.out.println("Enabling producer control flow...");
		byte[] msg = String.valueOf(true).getBytes();
		channel.basicPublish("flow.fx", "", null, msg);
		return true;
	}
	
	private boolean disableControlFlow(Channel channel) throws Exception {