<commit_before>/**
 * 
 */
package cn.strong.fastdfs.client;

import io.netty.channel.Channel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.pool.FixedChannelPool;
import io.netty.util.concurrent.Future;

import java.io.Closeable;
import java.io.IOException;
import java.net.InetSocketAddress;

import javax.annotation.PreDestroy;

import rx.Observable;

import cn.strong.fastdfs.client.protocol.request.Request;
import cn.strong.fastdfs.client.protocol.response.Receiver;
import cn.strong.fastdfs.ex.FastdfsException;

/**
 * FastDFS 连接交互模板类
 * 
 * @author liulongbiao
 *
 */
public class FastdfsTemplate implements Closeable {

	final NioEventLoopGroup group;
	final FastdfsChannelPoolMap poolMap;

	public FastdfsTemplate(Settings settings) {
		if (settings == null) {
			settings = new Settings();
		}
		group = new NioEventLoopGroup(settings.getEventLoopThreads());
		poolMap = new FastdfsChannelPoolMap(group, settings);
	}

	/**
	 * 访问 Fastdfs 服务器
	 * 
	 * @param addr
	 * @param sender
	 * @param decoder
	 * @return
	 */
	public <T> Observable<T> execute(InetSocketAddress addr, Request request, Receiver<T> receiver) {
		FixedChannelPool pool = poolMap.get(addr);
		pool.acquire().addListener((Future<Channel> f) -> {
			if (f.isCancelled()) {
				receiver.tryError(new FastdfsException("connection is canceled"));
			} else if (!f.isSuccess()) {
				receiver.tryError(f.cause());
			} else {
				Channel ch = f.getNow();
				receiver.observable().doAfterTerminate(() -> {
					pool.release(ch);
				});
				try {
					ch.attr(Receiver.RECEIVER).set(receiver);
					ch.writeAndFlush(request);
				} catch (Exception e) {
					receiver.tryError(e);
				}
			}
		});
		return receiver.observable();
	}

	@PreDestroy
	public void close() throws IOException {
		if (poolMap != null) {
			try {
				poolMap.close();
			} catch (Exception e) {
				// ignore
			}
		}
		if (group != null) {
			group.shutdownGracefully();
		}
	}
}<commit_msg>subscribe to doAfterTerminate<commit_after>/**
 * 
 */
package cn.strong.fastdfs.client;

import io.netty.channel.Channel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.pool.FixedChannelPool;
import io.netty.util.concurrent.Future;

import java.io.Closeable;
import java.io.IOException;
import java.net.InetSocketAddress;

import javax.annotation.PreDestroy;

import rx.Observable;

import cn.strong.fastdfs.client.protocol.request.Request;
import cn.strong.fastdfs.client.protocol.response.Receiver;
import cn.strong.fastdfs.ex.FastdfsException;

/**
 * FastDFS 连接交互模板类
 * 
 * @author liulongbiao
 *
 */
public class FastdfsTemplate implements Closeable {

	final NioEventLoopGroup group;
	final FastdfsChannelPoolMap poolMap;

	public FastdfsTemplate(Settings settings) {
		if (settings == null) {
			settings = new Settings();
		}
		group = new NioEventLoopGroup(settings.getEventLoopThreads());
		poolMap = new FastdfsChannelPoolMap(group, settings);
	}

	/**
	 * 访问 Fastdfs 服务器
	 * 
	 * @param addr
	 * @param sender
	 * @param decoder
	 * @return
	 */
	public <T> Observable<T> execute(InetSocketAddress addr, Request request, Receiver<T> receiver) {
		FixedChannelPool pool = poolMap.get(addr);
		pool.acquire().addListener((Future<Channel> f) -> {
			if (f.isCancelled()) {
				receiver.tryError(new FastdfsException("connection is canceled"));
			} else if (!f.isSuccess()) {
				receiver.tryError(f.cause());
			} else {
				Channel ch = f.getNow();
				receiver.observable().doAfterTerminate(() -> {
					pool.release(ch);
				}).subscribe();
				try {
					ch.attr(Receiver.RECEIVER).set(receiver);
					ch.writeAndFlush(request);
				} catch (Exception e) {
					receiver.tryError(e);
				}
			}
		});
		return receiver.observable();
	}

	@PreDestroy
	public void close() throws IOException {
		if (poolMap != null) {
			try {
				poolMap.close();
			} catch (Exception e) {
				// ignore
			}
		}
		if (group != null) {
			group.shutdownGracefully();
		}
	}
}
{'fraction_non_alphanumeric': 0.13432459677419356, 'fraction_numerical': 0.0, 'mean_word_length': 3.7819277108433735, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 15, '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': '6034426', 'n_tokens_mistral': 1528, 'n_tokens_neox': 1348, 'n_words': 255}