<commit_before>/**
 * 
 */
package com.trendrr.strest.server.connections;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentSkipListSet;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelFutureListener;
import org.jboss.netty.handler.codec.http.HttpRequest;
import org.jboss.netty.handler.codec.http.HttpResponse;

import com.trendrr.oss.DynMap;
import com.trendrr.strest.StrestUtil;
import com.trendrr.strest.server.ResponseBuilder;
import com.trendrr.strest.server.callbacks.DisconnectCallback;


/**
 * @author <NAME>
 * @created Jan 12, 2011
 * 
 */
public class StrestConnectionChannel implements Comparable<StrestConnectionChannel>{

	protected Log log = LogFactory.getLog(StrestConnectionChannel.class);
	
	Channel channel;
	
	private ConcurrentHashMap<String, StrestConnectionTxn> transactions = new ConcurrentHashMap<String, StrestConnectionTxn>();
	private ConcurrentHashMap<String,Object> connectionStorage = new ConcurrentHashMap<String,Object>();
	private ConcurrentLinkedQueue<DisconnectCallback> disconnectCallbacks = new ConcurrentLinkedQueue<DisconnectCallback>();
	
	public StrestConnectionChannel(Channel channel) {
		this.channel = channel;
	}
	
	/**
	 * registers a callback for when the connection is disconnected..
	 * @param callback
	 */
	public void onDisconnect(DisconnectCallback callback) {
		this.disconnectCallbacks.add(callback);
	}
	
	public Channel getChannel() {
		return this.channel;
	}

	/**
	 * gets the threadsafe connection storage, never null.
	 * @return
	 */
	public Map<String, Object> getStorage() {
		return connectionStorage;
	}

	public synchronized boolean isConnected() {
		return this.channel != null;
	}
	
	public synchronized void cleanup() {		
		//first cleanup all the txns
		for (String txnId : this.transactions.keySet()) {
			this.transactions.get(txnId).cleanup();
		}
		this.transactions = null;
		
		//call all the disconnectCallbacks
		while(!this.disconnectCallbacks.isEmpty()) {
			DisconnectCallback cb = this.disconnectCallbacks.poll();
			cb.disconnected(this);
		}
		this.disconnectCallbacks = null;
		this.channel = null;
		this.connectionStorage.clear();
	}
	
	public ChannelFuture sendMessage(ResponseBuilder responseBuilder) {
		return this.sendMessage(responseBuilder.getResponse());
	}
	
	public synchronized ChannelFuture sendMessage(HttpResponse response) {
		if (channel == null || !channel.isOpen()) {
			log.info("channel is closed, user has disconnected");
			return null;
		}
        
		// Write the response.
		if (StrestUtil.HEADERS.TXN_STATUS_VALUES.COMPLETE.equalsIgnoreCase(response.getHeader(StrestUtil.HEADERS.TXN_STATUS))) {
			//remove the txn
			this.txnComplete(response.getHeader(StrestUtil.HEADERS.TXN_ID));
		}
		return channel.write(response);
	}
	
	public void incoming(HttpRequest request) {<commit_msg>don't null out the channel so we don't accidentally lose it since it keys in the router<commit_after>/**
 * 
 */
package com.trendrr.strest.server.connections;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentSkipListSet;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelFutureListener;
import org.jboss.netty.handler.codec.http.HttpRequest;
import org.jboss.netty.handler.codec.http.HttpResponse;

import com.trendrr.oss.DynMap;
import com.trendrr.strest.StrestUtil;
import com.trendrr.strest.server.ResponseBuilder;
import com.trendrr.strest.server.callbacks.DisconnectCallback;


/**
 * @author <NAME>
 * @created Jan 12, 2011
 * 
 */
public class StrestConnectionChannel implements Comparable<StrestConnectionChannel>{

	protected Log log = LogFactory.getLog(StrestConnectionChannel.class);
	
	Channel channel;
	
	private ConcurrentHashMap<String, StrestConnectionTxn> transactions = new ConcurrentHashMap<String, StrestConnectionTxn>();
	private ConcurrentHashMap<String,Object> connectionStorage = new ConcurrentHashMap<String,Object>();
	private ConcurrentLinkedQueue<DisconnectCallback> disconnectCallbacks = new ConcurrentLinkedQueue<DisconnectCallback>();
	
	public StrestConnectionChannel(Channel channel) {
		this.channel = channel;
	}
	
	/**
	 * registers a callback for when the connection is disconnected..
	 * @param callback
	 */
	public void onDisconnect(DisconnectCallback callback) {
		this.disconnectCallbacks.add(callback);
	}
	
	public Channel getChannel() {
		return this.channel;
	}

	/**
	 * gets the threadsafe connection storage, never null.
	 * @return
	 */
	public Map<String, Object> getStorage() {
		return connectionStorage;
	}

	public synchronized boolean isConnected() {
		if (this.channel != null) {
			return this.channel.isConnected();
		}
		return false;
	}
	
	public synchronized void cleanup() {		
		//first cleanup all the txns
		for (String txnId : this.transactions.keySet()) {
			this.transactions.get(txnId).cleanup();
		}
		this.transactions = null;
		
		//call all the disconnectCallbacks
		while(!this.disconnectCallbacks.isEmpty()) {
			DisconnectCallback cb = this.disconnectCallbacks.poll();
			cb.disconnected(this);
		}
		this.disconnectCallbacks.clear();
//		this.channel = null;
		this.connectionStorage.clear();
	}
	
	public ChannelFuture sendMessage(ResponseBuilder responseBuilder) {
		return this.sendMessage(responseBuilder.getResponse());
	}
	
	public synchronized ChannelFuture sendMessage(HttpResponse response) {
		if (channel == null || !channel.isOpen()) {
			log.info("channel is closed, user has disconnected");
			return null;
		}
        
		// Write the response.
		if (StrestUtil.HEADERS.TXN_STATUS_VALUES.COMPLETE.equalsIgnoreCase(response.getHeader(StrestUtil.HEADERS.TXN_STATUS))) {
			//remove the txn
			this.txnComplete(response.getHeader(StrestUtil.HEADERS.TXN_ID));
		}
		return channel.write(response);
	}
	
	public void incoming(HttpRequest request) {
{'fraction_non_alphanumeric': 0.10595920968769917, 'fraction_numerical': 0.0019120458891013384, 'mean_word_length': 5.029779058597502, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '259650', 'n_tokens_mistral': 1995, 'n_tokens_neox': 1829, 'n_words': 383}