import backtype.storm.messaging.TaskMessage; //导入依赖的package包/类
/**
 * Enqueue a task message to be sent to server
 */
@Override
public void send(List<TaskMessage> messages) {
	// throw exception if the client is being closed
	if (isClosed()) {
		LOG.warn("Client is being closed, and does not take requests any more");
		return;
	}

	sendTimer.start();
	try {
		pushBatch(messages);
	} catch (Exception e) {
		throw new RuntimeException(e);
	} finally {
		sendTimer.stop();

	}
}
 

import backtype.storm.messaging.TaskMessage; //导入依赖的package包/类
@Override
public void onEvent(Object event, long sequence, boolean endOfBatch)
		throws Exception {
	if (event == null) {
		return;
	}

	TaskMessage message = (TaskMessage) event;

	MessageBatch messageBatch = messageBatchRef.getAndSet(null);
	if (null == messageBatch) {
		messageBatch = new MessageBatch(messageBatchSize);
	}

	messageBatch.add(message);

	if (messageBatch.isFull()) {
		batchQueue.offer(messageBatch);
	} else if (endOfBatch == true) {
		batchQueue.offer(messageBatch);
	} else {
		messageBatchRef.set(messageBatch);
	}
}
 

import backtype.storm.messaging.TaskMessage; //导入依赖的package包/类
/**
 * fetch a message from message queue synchronously (flags != 1) or
 * asynchronously (flags==1)
 */
@Override
public TaskMessage recv(int flags) {
	try {
		if ((flags & 0x01) == 0x01) {
			return (TaskMessage) recvQueue.poll();
			// non-blocking

		} else {
			return (TaskMessage) recvQueue.take();

		}

	} catch (Exception e) {
		LOG.warn("Occur unexception ", e);
		return null;
	}

}
 

import backtype.storm.messaging.TaskMessage; //导入依赖的package包/类
/**
 * create a buffer containing the encoding of this batch
 */
ChannelBuffer buffer() throws Exception {
	ChannelBufferOutputStream bout = new ChannelBufferOutputStream(
			ChannelBuffers.directBuffer(encoded_length));

	for (Object msg : msgs)
		if (msg instanceof TaskMessage)
			writeTaskMessage(bout, (TaskMessage) msg);
		else {
			// LOG.debug("Write one non-TaskMessage {}", msg );
			((ControlMessage) msg).write(bout);
		}

	// add a END_OF_BATCH indicator
	ControlMessage.EOB_MESSAGE.write(bout);
	// LOG.debug("ControlMessage.EOB_MESSAGE " );

	bout.close();

	return bout.buffer();
}
 

import backtype.storm.messaging.TaskMessage; //导入依赖的package包/类
/**
 * write a TaskMessage into a stream
 * 
 * Each TaskMessage is encoded as: task ... short(2) len ... int(4) payload
 * ... byte[] *
 */
private void writeTaskMessage(ChannelBufferOutputStream bout,
		TaskMessage message) throws Exception {
	int payload_len = 0;
	if (message.message() != null)
		payload_len = message.message().length;

	int task_id = message.task();
	if (task_id > Short.MAX_VALUE)
		throw new RuntimeException("Task ID should not exceed "
				+ Short.MAX_VALUE);

	bout.writeShort((short) task_id);
	bout.writeInt(payload_len);
	if (payload_len > 0)
		bout.write(message.message());

	// @@@ TESTING CODE
	// LOG.info("Write one message taskid:{}, len:{}, data:{}", task_id
	// , payload_len, JStormUtils.toPrintableString(message.message()) );
}
 

import backtype.storm.messaging.TaskMessage; //导入依赖的package包/类
@Override
public void run() {
	LOG.info("Successfully start ZMQ Recv thread");
	
	while(isClosed() == false) {
		try {
			TaskMessage message = recv(0);
			enqueue(message);
		}catch (Exception e) {
			LOG.warn("ZMQ Recv thread receive error", e);
		}
	}
	
	LOG.info("Successfully shutdown ZMQ Recv thread");
	
}
 

import backtype.storm.messaging.TaskMessage; //导入依赖的package包/类
@Override
public void handleEvent(Object event, boolean endOfBatch)
		throws Exception {
	TaskMessage message = (TaskMessage) event;

	int task = message.task();

	DisruptorQueue queue = deserializeQueues.get(task);
	if (queue == null) {
		LOG.warn("Received invalid message directed at port " + task
				+ ". Dropping...");
		return;
	}

	queue.publish(message.message());

}
 

import backtype.storm.messaging.TaskMessage; //导入依赖的package包/类
protected void pullTuples(Object event) {
    TupleExt tuple = (TupleExt) event;
    int taskid = tuple.getTargetTaskId();
    IConnection conn = getConnection(taskid);
    if (conn != null) {
        while (conn.available() == false) {
            try {
                Thread.sleep(1);
            } catch (InterruptedException e) {

            }
        }
        byte[] tupleMessage = serializer.serialize(tuple);
        TaskMessage taskMessage = new TaskMessage(taskid, tupleMessage);
        conn.send(taskMessage);
    }
}
 

import backtype.storm.messaging.TaskMessage; //导入依赖的package包/类
@Override
public synchronized void send(TaskMessage message) {
    // throw exception if the client is being closed
    if (isClosed()) {
        LOG.warn("Client is being closed, and does not take requests any more");
        return;
    }

    long start = System.nanoTime();
    try {
        pushBatch(message);
    } catch (Exception e) {
        throw new RuntimeException(e);
    } finally {
        long end = System.nanoTime();
        if (sendTimer != null) {
            sendTimer.update((end - start) / TimeUtils.NS_PER_US);
        }
    }
}
 

import backtype.storm.messaging.TaskMessage; //导入依赖的package包/类
@Override
public void onEvent(Object event, long sequence, boolean endOfBatch) throws Exception {
    if (event == null) {
        return;
    }

    TaskMessage message = (TaskMessage) event;

    MessageBatch messageBatch = messageBatchRef.getAndSet(null);
    if (null == messageBatch) {
        messageBatch = new MessageBatch(messageBatchSize);
    }

    messageBatch.add(message);

    if (messageBatch.isFull()) {
        batchQueue.offer(messageBatch);
    } else if (endOfBatch == true) {
        batchQueue.offer(messageBatch);
    } else {
        messageBatchRef.set(messageBatch);
    }
}
 

import backtype.storm.messaging.TaskMessage; //导入依赖的package包/类
void pushBatch(TaskMessage message) {

		if (TaskMessage.isEmpty(message)) {
			return;
		}

		MessageBatch messageBatch = messageBatchRef.getAndSet(null);
		if (null == messageBatch) {
			messageBatch = new MessageBatch(messageBatchSize);
		}

		messageBatch.add(message);

		Channel channel = isChannelReady();
		if (channel == null) {
			handleFailedChannel(messageBatch);
			return;
		}

		if (messageBatch.isFull()) {
			flushRequest(channel, messageBatch);

			return;
		}

		if (directlySend) {
			flushRequest(channel, messageBatch);
		} else {
			if(messageBatchRef.compareAndSet(null, messageBatch))
			    flush_later.set(true);
			else
			    LOG.error("MessageBatch will be lost. This should not happen.");
		}

		return;
	}
 

import backtype.storm.messaging.TaskMessage; //导入依赖的package包/类
/**
 * Enqueue a task message to be sent to server
 */
@Override
public void send(List<TaskMessage> messages) {
	for (TaskMessage msg : messages) {
		disruptorQueue.publish(msg);
	}
}
 

import backtype.storm.messaging.TaskMessage; //导入依赖的package包/类
@Override
public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) {
     
	Object msg = e.getMessage();
	if (msg == null)
		return;

	// end of batch?
	if (msg == ControlMessage.EOB_MESSAGE) {
		if (server.isSyncMode() == true) {
			Channel channel = ctx.getChannel();
			// simplify the logic, just send OK_RESPONSE
			channel.write(ControlMessage.OK_RESPONSE);
		}
		return;
	} else if (msg instanceof ControlMessage) {
		//LOG.debug("Receive ...{}", msg);
		return;
	}

	// enqueue the received message for processing
	try {
		server.enqueue((TaskMessage) msg);
	} catch (Exception e1) {
		LOG.warn("Failed to enqueue a request message" + e1.toString(), e);
		// Channel channel = ctx.getChannel();
		// incFailureCounter(channel);
	}
}
 

import backtype.storm.messaging.TaskMessage; //导入依赖的package包/类
private int msgEncodeLength(TaskMessage taskMsg) {
	if (taskMsg == null)
		return 0;

	int size = 6; // INT + SHORT
	if (taskMsg.message() != null)
		size += taskMsg.message().length;
	return size;
}
 

import backtype.storm.messaging.TaskMessage; //导入依赖的package包/类
@Override
public void send(List<TaskMessage> messages) {
	timer.start();

	try {
		for (TaskMessage message : messages) {
			ZeroMq.send(socket, message.message());
		}
	} finally {
		timer.stop();
		histogram.update(messages.size());
	}
}
 

import backtype.storm.messaging.TaskMessage; //导入依赖的package包/类
@Override
public TaskMessage recv(int flags) {
	byte[] data =  ZeroMq.recv(socket, flags);
	if (data == null || data.length <= 4) {
		return null;
	}
	int port = KryoTupleDeserializer.deserializeTaskId(data);
	return new TaskMessage(port, data);
}
 

import backtype.storm.messaging.TaskMessage; //导入依赖的package包/类
@Override
public void enqueue(TaskMessage message) {
	if (message != null ) {
		recvQueue.publish(message);
	}
	
}
 

import backtype.storm.messaging.TaskMessage; //导入依赖的package包/类
public BatchTupleRunable(WorkerData workerData) {
	super(workerData.getTransferQueue(), timer, BatchTupleRunable.class.getSimpleName(), workerData.getActive());
	this.sendingQueue = workerData.getSendingQueue();
	this.nodeportSocket = workerData.getNodeportSocket();
	this.taskNodeport = workerData.getTaskNodeport();
	this.dispatchMap = new HashMap<IConnection, List<TaskMessage>>();
	
	this.queue = workerData.getTransferQueue();
	Metrics.registerQueue(null, MetricDef.BATCH_TUPLE_QUEUE, this.queue, null, Metrics.MetricType.WORKER);
	
	this.queue.consumerStarted();
}
 

import backtype.storm.messaging.TaskMessage; //导入依赖的package包/类
public void handleFinish() {
	for (Entry<IConnection, List<TaskMessage>> entry: dispatchMap.entrySet()) {
		Pair<IConnection, List<TaskMessage>> pair = 
				new Pair<IConnection, List<TaskMessage>>(
						entry.getKey(), entry.getValue());
		
		sendingQueue.publish(pair);
	}
	
	dispatchMap.clear();
}
 

import backtype.storm.messaging.TaskMessage; //导入依赖的package包/类
@SuppressWarnings("unused")
@Override
public void handleEvent(Object event, boolean endOfBatch)
		throws Exception {

	handleOneEvent((TaskMessage)event);
	
	if (endOfBatch == true && isDirectSend == false) {
		handleFinish();
	}
	
}