| Method from org.apache.jk.common.ChannelSocket Detail: |
|---|
 public void accept(MsgContext ep) throws IOException {
if( sSocket==null ) return;
synchronized(this) {
while(paused) {
try{
wait();
} catch(InterruptedException ie) {
//Ignore, since can't happen
}
}
}
Socket s=sSocket.accept();
ep.setNote( socketNote, s );
if(log.isDebugEnabled() )
log.debug("Accepted socket " + s );
try {
setSocketOptions(s);
} catch(SocketException sex) {
log.debug("Error initializing Socket Options", sex);
}
requestCount++;
InputStream is=new BufferedInputStream(s.getInputStream());
OutputStream os;
if( bufferSize > 0 )
os = new BufferedOutputStream( s.getOutputStream(), bufferSize);
else
os = s.getOutputStream();
ep.setNote( isNote, is );
ep.setNote( osNote, os );
ep.setControl( tp );
}
|
void acceptConnections() {
if( log.isDebugEnabled() )
log.debug("Accepting ajp connections on " + port);
while( running ) {
try{
MsgContext ep=createMsgContext(packetSize);
ep.setSource(this);
ep.setWorkerEnv( wEnv );
this.accept(ep);
if( !running ) break;
// Since this is a long-running connection, we don't care
// about the small GC
SocketConnection ajpConn=
new SocketConnection(this, ep);
tp.runIt( ajpConn );
}catch(Exception ex) {
if (running)
log.warn("Exception executing accept" ,ex);
}
}
} Accept incoming connections, dispatch to the thread pool |
public void addNotificationListener(NotificationListener listener,
NotificationFilter filter,
Object handback) throws IllegalArgumentException {
if( nSupport==null ) nSupport=new NotificationBroadcasterSupport();
nSupport.addNotificationListener(listener, filter, handback);
} |
public void close(MsgContext ep) throws IOException {
Socket s=(Socket)ep.getNote( socketNote );
s.close();
} |
public void destroy() throws IOException {
running = false;
try {
/* If we disabled the channel return */
if (port == 0)
return;
tp.shutdown();
if(!paused) {
unLockSocket();
}
sSocket.close(); // XXX?
if( tpOName != null ) {
Registry.getRegistry(null, null).unregisterComponent(tpOName);
}
if( rgOName != null ) {
Registry.getRegistry(null, null).unregisterComponent(rgOName);
}
} catch(Exception e) {
log.info("Error shutting down the channel " + port + " " +
e.toString());
if( log.isDebugEnabled() ) log.debug("Trace", e);
}
} |
public int flush(Msg msg,
MsgContext ep) throws IOException {
if( bufferSize > 0 ) {
OutputStream os=(OutputStream)ep.getNote( osNote );
os.flush();
}
return 0;
} |
public String getAddress() {
if( inet!=null)
return inet.toString();
return "/0.0.0.0";
} |
public int getBufferSize() {
return bufferSize;
} |
public String getChannelName() {
String encodedAddr = "";
if (inet != null && !"0.0.0.0".equals(inet.getHostAddress())) {
encodedAddr = getAddress();
if (encodedAddr.startsWith("/"))
encodedAddr = encodedAddr.substring(1);
encodedAddr = URLEncoder.encode(encodedAddr) + "-";
}
return ("jk-" + encodedAddr + port);
} |
public boolean getDaemon() {
return tp.getDaemon();
} |
public int getInstanceId() {
return port-startPort;
} At startup we'll look for the first free port in the range.
The difference between this port and the beggining of the range
is the 'id'.
This is usefull for lb cases ( less config ). |
public int getMaxPort() {
return maxPort;
} |
public int getMaxSpareThreads() {
return tp.getMaxSpareThreads();
} |
public int getMaxThreads() {
return tp.getMaxThreads();
} |
public int getMinSpareThreads() {
return tp.getMinSpareThreads();
} |
public MBeanNotificationInfo[] getNotificationInfo() {
return notifInfo;
} |
public int getPacketSize() {
return packetSize;
} |
public int getPort() {
return port;
} |
public long getRequestCount() {
return requestCount;
} |
public int getServerTimeout() {
return serverTimeout;
} |
public int getSoLinger() {
return linger;
} |
public int getSoTimeout() {
return socketTimeout;
} |
public boolean getTcpNoDelay() {
return tcpNoDelay;
} |
public ThreadPool getThreadPool() {
return tp;
} |
public void init() throws IOException {
// Find a port.
if (startPort == 0) {
port = 0;
if(log.isInfoEnabled())
log.info("JK: ajp13 disabling channelSocket");
running = true;
return;
}
if (maxPort < startPort)
maxPort = startPort;
for( int i=startPort; i< =maxPort; i++ ) {
try {
if( inet == null ) {
sSocket = new ServerSocket( i, 0 );
} else {
sSocket=new ServerSocket( i, 0, inet );
}
port=i;
break;
} catch( IOException ex ) {
if(log.isInfoEnabled())
log.info("Port busy " + i + " " + ex.toString());
continue;
}
}
if( sSocket==null ) {
log.error("Can't find free port " + startPort + " " + maxPort );
return;
}
if(log.isInfoEnabled())
log.info("JK: ajp13 listening on " + getAddress() + ":" + port );
// If this is not the base port and we are the 'main' channleSocket and
// SHM didn't already set the localId - we'll set the instance id
if( "channelSocket".equals( name ) &&
port != startPort &&
(wEnv.getLocalId()==0) ) {
wEnv.setLocalId( port - startPort );
}
if( serverTimeout > 0 )
sSocket.setSoTimeout( serverTimeout );
// XXX Reverse it - > this is a notification generator !!
if( next==null && wEnv!=null ) {
if( nextName!=null )
setNext( wEnv.getHandler( nextName ) );
if( next==null )
next=wEnv.getHandler( "dispatch" );
if( next==null )
next=wEnv.getHandler( "request" );
}
JMXRequestNote =wEnv.getNoteId( WorkerEnv.ENDPOINT_NOTE, "requestNote");
running = true;
// Run a thread that will accept connections.
// XXX Try to find a thread first - not sure how...
if( this.domain != null ) {
try {
tpOName=new ObjectName(domain + ":type=ThreadPool,name=" +
getChannelName());
Registry.getRegistry(null, null)
.registerComponent(tp, tpOName, null);
rgOName = new ObjectName
(domain+":type=GlobalRequestProcessor,name=" + getChannelName());
Registry.getRegistry(null, null)
.registerComponent(global, rgOName, null);
} catch (Exception e) {
log.error("Can't register threadpool" );
}
}
tp.start();
SocketAcceptor acceptAjp=new SocketAcceptor( this );
tp.runIt( acceptAjp);
} |
public int invoke(Msg msg,
MsgContext ep) throws IOException {
int type=ep.getType();
switch( type ) {
case JkHandler.HANDLE_RECEIVE_PACKET:
if( log.isDebugEnabled()) log.debug("RECEIVE_PACKET ?? ");
return receive( msg, ep );
case JkHandler.HANDLE_SEND_PACKET:
return send( msg, ep );
case JkHandler.HANDLE_FLUSH:
return flush( msg, ep );
}
if( log.isDebugEnabled() )
log.debug("Call next " + type + " " + next);
// Send notification
if( nSupport!=null ) {
Notification notif=(Notification)ep.getNote(notifNote);
if( notif==null ) {
notif=new Notification("channelSocket.message", ep, requestCount );
ep.setNote( notifNote, notif);
}
nSupport.sendNotification(notif);
}
if( next != null ) {
return next.invoke( msg, ep );
} else {
log.info("No next ");
}
return OK;
} |
public boolean isSameAddress(MsgContext ep) {
Socket s=(Socket)ep.getNote( socketNote );
return isSameAddress( s.getLocalAddress(), s.getInetAddress());
} |
public static boolean isSameAddress(InetAddress server,
InetAddress client) {
// Compare the byte array versions of the two addresses
byte serverAddr[] = server.getAddress();
byte clientAddr[] = client.getAddress();
if (serverAddr.length != clientAddr.length)
return (false);
boolean match = true;
for (int i = 0; i < serverAddr.length; i++) {
if (serverAddr[i] != clientAddr[i]) {
match = false;
break;
}
}
if (match)
return (true);
// Compare the reversed form of the two addresses
for (int i = 0; i < serverAddr.length; i++) {
if (serverAddr[i] != clientAddr[(serverAddr.length-1)-i])
return (false);
}
return (true);
} Return true if the specified client and server addresses
are the same. This method works around a bug in the IBM 1.1.8 JVM on
Linux, where the address bytes are returned reversed in some
circumstances. |
public void open(MsgContext ep) throws IOException {
} |
public void pause() throws Exception {
synchronized(this) {
paused = true;
unLockSocket();
}
} |
void processConnection(MsgContext ep) {
try {
MsgAjp recv=new MsgAjp(packetSize);
while( running ) {
if(paused) { // Drop the connection on pause
break;
}
int status= this.receive( recv, ep );
if( status < = 0 ) {
if( status==-3)
log.debug( "server has been restarted or reset this connection" );
else
log.warn("Closing ajp connection " + status );
break;
}
ep.setLong( MsgContext.TIMER_RECEIVED, System.currentTimeMillis());
ep.setType( 0 );
// Will call next
status= this.invoke( recv, ep );
if( status!= JkHandler.OK ) {
log.warn("processCallbacks status " + status );
ep.action(ActionCode.ACTION_CLOSE, ep.getRequest().getResponse());
break;
}
}
} catch( Exception ex ) {
String msg = ex.getMessage();
if( msg != null && msg.indexOf( "Connection reset" ) >= 0)
log.debug( "Server has been restarted or reset this connection");
else if (msg != null && msg.indexOf( "Read timed out" ) >=0 )
log.debug( "connection timeout reached");
else
log.error( "Error, processing connection", ex);
} finally {
/*
* Whatever happened to this connection (remote closed it, timeout, read error)
* the socket SHOULD be closed, or we may be in situation where the webserver
* will continue to think the socket is still open and will forward request
* to tomcat without receiving ever a reply
*/
try {
this.close( ep );
}
catch( Exception e) {
log.error( "Error, closing connection", e);
}
try{
Request req = (Request)ep.getRequest();
if( req != null ) {
ObjectName roname = (ObjectName)ep.getNote(JMXRequestNote);
if( roname != null ) {
Registry.getRegistry(null, null).unregisterComponent(roname);
}
req.getRequestProcessor().setGlobalProcessor(null);
}
} catch( Exception ee) {
log.error( "Error, releasing connection",ee);
}
}
} Process a single ajp connection. |
public int read(MsgContext ep,
byte[] b,
int offset,
int len) throws IOException {
InputStream is=(InputStream)ep.getNote( isNote );
int pos = 0;
int got;
while(pos < len) {
try {
got = is.read(b, pos + offset, len - pos);
} catch(SocketException sex) {
if(pos > 0) {
log.info("Error reading data after "+pos+"bytes",sex);
} else {
log.debug("Error reading data", sex);
}
got = -1;
}
if (log.isTraceEnabled()) {
log.trace("read() " + b + " " + (b==null ? 0: b.length) + " " +
offset + " " + len + " = " + got );
}
// connection just closed by remote.
if (got < = 0) {
// This happens periodically, as apache restarts
// periodically.
// It should be more gracefull ! - another feature for Ajp14
// log.warn( "server has closed the current connection (-1)" );
return -3;
}
pos += got;
}
return pos;
} Read N bytes from the InputStream, and ensure we got them all
Under heavy load we could experience many fragmented packets
just read Unix Network Programming to recall that a call to
read didn't ensure you got all the data you want
from read() Linux manual
On success, the number of bytes read is returned (zero indicates end
of file),and the file position is advanced by this number.
It is not an error if this number is smaller than the number of bytes
requested; this may happen for example because fewer bytes
are actually available right now (maybe because we were close to
end-of-file, or because we are reading from a pipe, or from a
terminal), or because read() was interrupted by a signal.
On error, -1 is returned, and errno is set appropriately. In this
case it is left unspecified whether the file position (if any) changes. |
public int receive(Msg msg,
MsgContext ep) throws IOException {
if (log.isDebugEnabled()) {
log.debug("receive() ");
}
byte buf[]=msg.getBuffer();
int hlen=msg.getHeaderLength();
// XXX If the length in the packet header doesn't agree with the
// actual number of bytes read, it should probably return an error
// value. Also, callers of this method never use the length
// returned -- should probably return true/false instead.
int rd = this.read(ep, buf, 0, hlen );
if(rd < 0) {
// Most likely normal apache restart.
// log.warn("Wrong message " + rd );
return rd;
}
msg.processHeader();
/* After processing the header we know the body
length
*/
int blen=msg.getLen();
// XXX check if enough space - it's assert()-ed !!!
int total_read = 0;
total_read = this.read(ep, buf, hlen, blen);
if ((total_read < = 0) && (blen > 0)) {
log.warn("can't read body, waited #" + blen);
return -1;
}
if (total_read != blen) {
log.warn( "incomplete read, waited #" + blen +
" got only " + total_read);
return -2;
}
return total_read;
} |
public void registerRequest(Request req,
MsgContext ep,
int count) {
if(this.domain != null) {
try {
RequestInfo rp=req.getRequestProcessor();
rp.setGlobalProcessor(global);
ObjectName roname = new ObjectName
(getDomain() + ":type=RequestProcessor,worker="+
getChannelName()+",name=JkRequest" +count);
ep.setNote(JMXRequestNote, roname);
Registry.getRegistry(null, null).registerComponent( rp, roname, null);
} catch( Exception ex ) {
log.warn("Error registering request");
}
}
} |
public void reinit() throws IOException {
destroy();
init();
} Called after you change some fields at runtime using jmx.
Experimental for now. |
public void removeNotificationListener(NotificationListener listener) throws ListenerNotFoundException {
if( nSupport!=null)
nSupport.removeNotificationListener(listener);
} |
public void resetCounters() {
requestCount=0;
} |
public void resume() throws Exception {
synchronized(this) {
paused = false;
notify();
}
} |
public int send(Msg msg,
MsgContext ep) throws IOException {
msg.end(); // Write the packet header
byte buf[]=msg.getBuffer();
int len=msg.getLen();
if(log.isTraceEnabled() )
log.trace("send() " + len + " " + buf[4] );
OutputStream os=(OutputStream)ep.getNote( osNote );
os.write( buf, 0, len );
return len;
} |
public void sendNewMessageNotification(Notification notification) {
if( nSupport!= null )
nSupport.sendNotification(notification);
} |
public void setAddress(InetAddress inet) {
this.inet=inet;
} |
public void setAddress(String inet) {
try {
this.inet= InetAddress.getByName( inet );
} catch( Exception ex ) {
log.error("Error parsing "+inet,ex);
}
} jmx:managed-attribute description="Bind on a specified address" access="READ_WRITE" |
public void setBacklog(int i) {
} |
public void setBufferSize(int bs) {
bufferSize = bs;
} |
public void setDaemon(boolean b) {
tp.setDaemon( b );
} If set to false, the thread pool will be created in
non-daemon mode, and will prevent main from exiting |
public void setMaxPort(int i) {
maxPort=i;
} |
public void setMaxSpareThreads(int i) {
if( log.isDebugEnabled()) log.debug("Setting maxSpareThreads " + i);
tp.setMaxSpareThreads(i);
} |
public void setMaxThreads(int i) {
if( log.isDebugEnabled()) log.debug("Setting maxThreads " + i);
tp.setMaxThreads(i);
} |
public void setMinSpareThreads(int i) {
if( log.isDebugEnabled()) log.debug("Setting minSpareThreads " + i);
tp.setMinSpareThreads(i);
} |
public void setNotificationInfo(MBeanNotificationInfo[] info) {
this.notifInfo=info;
} |
public void setPacketSize(int ps) {
if(ps < AjpConstants.MAX_PACKET_SIZE) {
ps = AjpConstants.MAX_PACKET_SIZE;
}
packetSize = ps;
} |
public void setPort(int port) {
this.startPort=port;
this.port=port;
this.maxPort=port+10;
} Set the port for the ajp13 channel.
To support seemless load balancing and jni, we treat this
as the 'base' port - we'll try up until we find one that is not
used. We'll also provide the 'difference' to the main coyote
handler - that will be our 'sessionID' and the position in
the scoreboard and the suffix for the unix domain socket.
jmx:managed-attribute description="Port to listen" access="READ_WRITE" |
public void setServerTimeout(int timeout) {
this.serverTimeout = timeout;
} |
public void setSoLinger(int i) {
linger=i;
} |
public void setSoTimeout(int i) {
socketTimeout=i;
} |
public void setTcpNoDelay(boolean b) {
tcpNoDelay=b;
} |
public void start() throws IOException {
if( sSocket==null )
init();
} |
public void stop() throws IOException {
destroy();
} |
org.apache.jk.core.JkHandler
Save This Page