BioReplicationTask.java
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.catalina.tribes.transport.bio;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
import org.apache.catalina.tribes.ChannelMessage;
import org.apache.catalina.tribes.io.BufferPool;
import org.apache.catalina.tribes.io.ChannelData;
import org.apache.catalina.tribes.io.ListenCallback;
import org.apache.catalina.tribes.io.ObjectReader;
import org.apache.catalina.tribes.transport.AbstractRxTask;
import org.apache.catalina.tribes.transport.Constants;
import org.apache.catalina.tribes.util.StringManager;
import org.apache.juli.logging.Log;
import org.apache.juli.logging.LogFactory;
/**
* A worker thread class which can drain channels and echo-back the input. Each instance is constructed with a reference
* to the owning thread pool object. When started, the thread loops forever waiting to be awakened to service the
* channel associated with a SelectionKey object. The worker is tasked by calling its serviceChannel() method with a
* SelectionKey object. The serviceChannel() method stores the key reference in the thread object then calls notify() to
* wake it up. When the channel has been drained, the worker thread returns itself to its parent pool.
*
* @deprecated This will be removed in Tomcat 10
*/
@Deprecated
public class BioReplicationTask extends AbstractRxTask {
private static final Log log = LogFactory.getLog(BioReplicationTask.class);
protected static final StringManager sm = StringManager.getManager(BioReplicationTask.class);
protected Socket socket;
protected ObjectReader reader;
public BioReplicationTask(ListenCallback callback) {
super(callback);
}
// loop forever waiting for work to do
@Override
public synchronized void run() {
if (socket == null) {
return;
}
try {
drainSocket();
} catch (Exception x) {
log.error(sm.getString("bioReplicationTask.unable.service"), x);
} finally {
try {
socket.close();
} catch (Exception e) {
if (log.isDebugEnabled()) {
log.debug(sm.getString("bioReplicationTask.socket.closeFailed"), e);
}
}
try {
reader.close();
} catch (Exception e) {
if (log.isDebugEnabled()) {
log.debug(sm.getString("bioReplicationTask.reader.closeFailed"), e);
}
}
reader = null;
socket = null;
}
// done, ready for more, return to pool
if (getTaskPool() != null) {
getTaskPool().returnWorker(this);
}
}
public synchronized void serviceSocket(Socket socket, ObjectReader reader) {
this.socket = socket;
this.reader = reader;
}
protected void execute(ObjectReader reader) throws Exception {
int pkgcnt = reader.count();
if (pkgcnt > 0) {
ChannelMessage[] msgs = reader.execute();
for (int i = 0; i < msgs.length; i++) {
/*
* Use send ack here if you want to ack the request to the remote server before completing the request
* This is considered an asynchronous request
*/
if (ChannelData.sendAckAsync(msgs[i].getOptions())) {
sendAck(Constants.ACK_COMMAND);
}
try {
// process the message
getCallback().messageDataReceived(msgs[i]);
/*
* Use send ack here if you want the request to complete on this server before sending the ack to
* the remote server This is considered a synchronized request
*/
if (ChannelData.sendAckSync(msgs[i].getOptions())) {
sendAck(Constants.ACK_COMMAND);
}
} catch (Exception x) {
if (ChannelData.sendAckSync(msgs[i].getOptions())) {
sendAck(Constants.FAIL_ACK_COMMAND);
}
log.error(sm.getString("bioReplicationTask.messageDataReceived.error"), x);
}
if (getUseBufferPool()) {
BufferPool.getBufferPool().returnBuffer(msgs[i].getMessage());
msgs[i].setMessage(null);
}
}
}
}
/**
* The actual code which drains the channel associated with the given key. This method assumes the key has been
* modified prior to invocation to turn off selection interest in OP_READ. When this method completes it re-enables
* OP_READ and calls wakeup() on the selector so the selector will resume watching this channel.
*
* @throws Exception IO exception or execute exception
*/
protected void drainSocket() throws Exception {
InputStream in = socket.getInputStream();
// loop while data available, channel is non-blocking
byte[] buf = new byte[1024];
int length = in.read(buf);
while (length >= 0) {
int count = reader.append(buf, 0, length, true);
if (count > 0) {
execute(reader);
}
length = in.read(buf);
}
}
/**
* Send a reply-acknowledgment (6,2,3)
*
* @param command The command to write
*/
protected void sendAck(byte[] command) {
try {
OutputStream out = socket.getOutputStream();
out.write(command);
out.flush();
if (log.isTraceEnabled()) {
log.trace("ACK sent to " + socket.getPort());
}
} catch (java.io.IOException x) {
log.warn(sm.getString("bioReplicationTask.unable.sendAck", x.getMessage()));
}
}
@Override
public void close() {
try {
socket.close();
} catch (Exception e) {
if (log.isDebugEnabled()) {
log.debug(sm.getString("bioReplicationTask.socket.closeFailed"), e);
}
}
try {
reader.close();
} catch (Exception e) {
if (log.isDebugEnabled()) {
log.debug(sm.getString("bioReplicationTask.reader.closeFailed"), e);
}
}
reader = null;
socket = null;
super.close();
}
}