package com.xrtb.bidder;

import java.io.File;
import java.util.concurrent.ConcurrentLinkedQueue;

import org.redisson.RedissonClient;
import org.redisson.core.RTopic;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.xrtb.tools.logmaster.AppendToFile;
import com.xrtb.tools.logmaster.FileLogger;
import com.xrtb.tools.logmaster.Spark;

/**
 * A publisher for REDIS based messages, sharable by multiple threads.
 * 
 * @author Ben M. Faul
 *
 */
public class Publisher implements Runnable {
	/** The objects thread */
	Thread me;
	/** The JEDIS connection used */
	String channel;
	/** The topic of messages */
	RTopic logger;
	/** The queue of messages */
	ConcurrentLinkedQueue queue = new ConcurrentLinkedQueue();

	/** Filename, if not using redis */
	String fileName;
	StringBuilder sb;
	ObjectMapper mapper;
	boolean errored = false;

	/**
	 * Constructor for base class.
	 * 
	 * @param conn
	 *            Jedis. The REDIS connection.
	 * @param channel
	 *            String. The topic name to publish on.
	 * @throws Exception. Throws
	 *             exceptions on REDIS errors
	 */
	public Publisher(RedissonClient redisson, String channel) throws Exception {
		this.channel = channel;
		logger = redisson.getTopic(channel);
		me = new Thread(this);
		me.start();

	}

	public Publisher(String fileName) throws Exception {
		int i = fileName.indexOf("file://");
		if (i > -1) {
			fileName = fileName.substring(7);
		}
		this.fileName = fileName;
		mapper = new ObjectMapper();
		sb = new StringBuilder();
		me = new Thread(this);
		me.start();
	}

	/**
	 * Return the publishing channel
	 * 
	 * @return RTopic. The RTopic channel.
	 */
	public RTopic getChannel() {
		return logger;
	}

	@Override
	public void run() {
		if (logger == null)
			runFileLogger();
		else
			runRedisLogger();
	}

	void runFileLogger() {
		Object obj = null;

		while (true) {
			try {
				Thread.sleep(60000);

				synchronized (sb) {
					if (sb.length() != 0) {
						AppendToFile.item(fileName, sb);
						sb.setLength(0);
					}
				}
			} catch (Exception error) {
				errored = true;
				try {
					Controller.getInstance().sendLog(1, "Publisher:"+fileName,"Publisher log error on " + fileName + ", error = " + error.toString());
				} catch (Exception e) {
				}
				error.printStackTrace();
				sb.setLength(0);
			}
		}
	}

	void runRedisLogger() {
		String str = null;
		Object msg = null;
		while (true) {
			try {
				if ((msg = queue.poll()) != null) {
					// System.out.println("message");
					logger.publishAsync(msg);
				}
				Thread.sleep(1);
			} catch (Exception e) {
				e.printStackTrace();
				// return;
			}
		}
	}

	/**
	 * Out of band write, like when you absolutely have to send a notice now
	 * (Like a shutdown notice)
	 * 
	 * @param Object
	 */
	public void writeFast(Object msg) {
		logger.publishAsync(msg);
	}

	/**
	 * Add a message to the messages queue.
	 * 
	 * @param s
	 *            . String. JSON formatted message.
	 */
	public void add(Object s) {
		if (fileName != null) {
			if (errored)
				return;
			
			synchronized (sb) {
				sb.append(s);
				sb.append("\n");
			}
		} else
			queue.add(s);
	}
}
@@ -526,16 +526,26 @@
 ead */%0A%09
+protected 
 Thread m
@@ -675,16 +675,26 @@
 ges */%0A%09
+protected 
 Concurre
@@ -779,24 +779,34 @@
 g redis */%0A%09
+protected 
 String fileN
@@ -811,16 +811,26 @@
 eName;%0A%09
+protected 
 StringBu
@@ -840,16 +840,26 @@
 er sb;%0A%09
+protected 
 ObjectMa
@@ -872,16 +872,26 @@
 apper;%0A%09
+protected 
 boolean 
@@ -908,16 +908,44 @@
 false;%0A%0A
+%09public Publisher() %7B%0A%09%09%0A%09%7D%0A
 %09/**%0A%09 *
@@ -1898,24 +1898,31 @@
 ger();%0A%09%7D%0A%0A%09
+public 
 void runFile
@@ -2429,16 +2429,59 @@
 %09%7D%0A%09%7D%0A%0A%09
+/**%0A%09 * Runs the REDIS logger%0A%09 */%0A%09public 
 void run

Change visibility on publisher so subclasess can use pieces