Merge remote-tracking branch 'origin/master' into jetty-8

Conflicts:
	VERSION.txt
	jetty-osgi/jetty-osgi-boot-logback/META-INF/MANIFEST.MF
	jetty-osgi/jetty-osgi-equinoxtools/src/main/java/org/eclipse/jetty/osgi/equinoxtools/WebEquinoxToolsActivator.java
	pom.xml
This commit is contained in:
Jan Bartel 2011-08-01 14:39:39 +10:00
commit 02a66a00fb
48 changed files with 4516 additions and 393 deletions

View File

@ -19,7 +19,12 @@ jetty-7.5.0-SNAPSHOT
+ 352421 HttpURI paths beginning with '.'
+ 352684 Implemented spinning thread analyzer
+ 352786 GzipFilter fails to pass parameters to GzipResponseWrapper
+ 353073 WebSocketClient
+ 353095 maven-jetty-plugin: PermGen leak due to javax.el.BeanELResolver
+ 353165 addJars can follow symbolic link jar files
+ 353210 Bundle-Version in o.e.j.o.boot.logback fix
+ 353465 JAASLoginService ignores callbackHandlerClass
jetty-7.4.4.v20110707 July 7th 2011
+ 308851 Converted all jetty-client module tests to JUnit 4
+ 345268 JDBCSessionManager does not work with maxInactiveInterval = -1

View File

@ -93,7 +93,13 @@ class SelectConnector extends AbstractLifeCycle implements HttpClient.Connector,
Address address = destination.isProxied() ? destination.getProxy() : destination.getAddress();
channel.socket().setTcpNoDelay(true);
if (!_httpClient.isConnectBlocking())
if (_httpClient.isConnectBlocking())
{
channel.socket().connect(address.toSocketAddress(), _httpClient.getConnectTimeout());
channel.configureBlocking(false);
_selectorManager.register( channel, destination );
}
else
{
channel.configureBlocking( false );
channel.connect(address.toSocketAddress());
@ -102,12 +108,6 @@ class SelectConnector extends AbstractLifeCycle implements HttpClient.Connector,
_httpClient.schedule(connectTimeout,_httpClient.getConnectTimeout());
_connectingChannels.put(channel, connectTimeout);
}
else
{
channel.socket().connect(address.toSocketAddress(), _httpClient.getConnectTimeout());
channel.configureBlocking(false);
_selectorManager.register( channel, destination );
}
}
catch(IOException ex)

View File

@ -103,6 +103,14 @@ public class WebSocketUpgradeTest
_results.add("clientWS.onMessage");
_results.add(data);
}
/* ------------------------------------------------------------ */
public void onError(String message, Throwable ex)
{
_results.add("clientWS.onError");
_results.add(message);
_results.add(ex);
}
};
@ -245,6 +253,14 @@ public class WebSocketUpgradeTest
_results.add(data);
}
/* ------------------------------------------------------------ */
public void onError(String message, Throwable ex)
{
_results.add("serverWS.onError");
_results.add(message);
_results.add(ex);
}
/* ------------------------------------------------------------ */
public void onClose(int code, String message)
{

View File

@ -59,6 +59,7 @@ import org.eclipse.jetty.util.log.Log;
public class HttpFields
{
/* ------------------------------------------------------------ */
public static final String __COOKIE_DELIM="\"\\\n\r\t\f\b%+ ;=";
public static final TimeZone __GMT = TimeZone.getTimeZone("GMT");
public static final BufferDateCache __dateCache = new BufferDateCache("EEE, dd MMM yyyy HH:mm:ss 'GMT'", Locale.US);
@ -966,7 +967,7 @@ public class HttpFields
final boolean isHttpOnly,
int version)
{
String delim=_maxCookieVersion==0?"":"\"\\\n\r\t\f\b%+ ;=";
String delim=_maxCookieVersion==0?"":__COOKIE_DELIM;
// Check arguments
if (name == null || name.length() == 0)

94
jetty-nosql/pom.xml Normal file
View File

@ -0,0 +1,94 @@
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
<parent>
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-project</artifactId>
<version>7.5.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>jetty-nosql</artifactId>
<name>Jetty :: NoSQL Session Managers</name>
<properties>
<jetty-version>${project.version}</jetty-version>
<junit4-version>${junit-version}</junit4-version>
<bundle-symbolic-name>${project.groupId}.mongodb</bundle-symbolic-name> </properties>
<build>
<defaultGoal>install</defaultGoal>
<plugins>
<plugin>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<source>1.6</source>
<target>1.6</target>
<verbose>false</verbose>
</configuration>
</plugin>
<plugin>
<groupId>org.apache.felix</groupId>
<artifactId>maven-bundle-plugin</artifactId>
<configuration>
<instructions>
<Import-Package>javax.servlet.*;version="[2.5,3.0)",org.eclipse.jetty.server.session.jmx;version="[7.5,8)";resolution:=optional,,org.eclipse.jetty.*;version="[7.5,8)",*</Import-Package>
</instructions>
</configuration>
<extensions>true</extensions>
<executions>
<execution>
<goals>
<goal>manifest</goal>
</goals>
</execution>
</executions>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-jar-plugin</artifactId>
<executions>
<execution>
<id>artifact-jar</id>
<goals>
<goal>jar</goal>
</goals>
</execution>
</executions>
<configuration>
<archive>
<manifestFile>${project.build.outputDirectory}/META-INF/MANIFEST.MF</manifestFile>
</archive>
</configuration>
</plugin>
</plugins>
</build>
<dependencies>
<dependency>
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-server</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>${junit4-version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-jmx</artifactId>
<version>${project.version}</version>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.mongodb</groupId>
<artifactId>mongo-java-driver</artifactId>
<version>2.6.1</version>
<type>jar</type>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.eclipse.jetty.tests</groupId>
<artifactId>test-sessions-common</artifactId>
<version>${project.version}</version>
<scope>test</scope>
</dependency>
</dependencies>
</project>

View File

@ -0,0 +1,175 @@
package org.eclipse.jetty.nosql;
//========================================================================
//Copyright (c) 2011 Intalio, Inc.
//------------------------------------------------------------------------
//All rights reserved. This program and the accompanying materials
//are made available under the terms of the Eclipse Public License v1.0
//and Apache License v2.0 which accompanies this distribution.
//The Eclipse Public License is available at
//http://www.eclipse.org/legal/epl-v10.html
//The Apache License v2.0 is available at
//http://www.opensource.org/licenses/apache2.0.php
//You may elect to redistribute this code under either of these licenses.
//========================================================================
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import org.eclipse.jetty.server.session.AbstractSession;
import org.eclipse.jetty.util.log.Log;
import org.eclipse.jetty.util.log.Logger;
/* ------------------------------------------------------------ */
public class NoSqlSession extends AbstractSession
{
private final static Logger __log = Log.getLogger("org.eclipse.jetty.server.session");
private final NoSqlSessionManager _manager;
private Set<String> _dirty;
private final AtomicInteger _active = new AtomicInteger();
private Object _version;
private long _lastSync;
/* ------------------------------------------------------------ */
public NoSqlSession(NoSqlSessionManager manager, long created, long accessed, String clusterId)
{
super(manager, created,accessed,clusterId);
_manager=manager;
save(true);
_active.incrementAndGet();
}
/* ------------------------------------------------------------ */
public NoSqlSession(NoSqlSessionManager manager, long created, long accessed, String clusterId, Object version)
{
super(manager, created,accessed,clusterId);
_manager=manager;
_version=version;
}
/* ------------------------------------------------------------ */
@Override
public Object doPutOrRemove(String name, Object value)
{
synchronized (this)
{
if (_dirty==null)
_dirty=new HashSet<String>();
_dirty.add(name);
Object old = super.doPutOrRemove(name,value);
if (_manager.getSavePeriod()==-2)
save(true);
return old;
}
}
/* ------------------------------------------------------------ */
@Override
protected void checkValid() throws IllegalStateException
{
super.checkValid();
}
/* ------------------------------------------------------------ */
@Override
protected boolean access(long time)
{
__log.debug("NoSqlSession:access:active "+_active);
if (_active.incrementAndGet()==1)
{
int period=_manager.getStalePeriod()*1000;
if (period==0)
refresh();
else if (period>0)
{
long stale=time-_lastSync;
__log.debug("NoSqlSession:access:stale "+stale);
if (stale>period)
refresh();
}
}
return super.access(time);
}
/* ------------------------------------------------------------ */
@Override
protected void complete()
{
super.complete();
if(_active.decrementAndGet()==0)
{
switch(_manager.getSavePeriod())
{
case 0:
save(isValid());
break;
case 1:
if (isDirty())
save(isValid());
break;
}
}
}
/* ------------------------------------------------------------ */
@Override
protected void doInvalidate() throws IllegalStateException
{
super.doInvalidate();
save(false);
}
/* ------------------------------------------------------------ */
protected void save(boolean activateAfterSave)
{
synchronized (this)
{
_version=_manager.save(this,_version,activateAfterSave);
_lastSync=getAccessed();
}
}
/* ------------------------------------------------------------ */
protected void refresh()
{
synchronized (this)
{
_version=_manager.refresh(this,_version);
}
}
/* ------------------------------------------------------------ */
public boolean isDirty()
{
synchronized (this)
{
return _dirty!=null && !_dirty.isEmpty();
}
}
/* ------------------------------------------------------------ */
public Set<String> takeDirty()
{
synchronized (this)
{
Set<String> dirty=_dirty;
if (dirty==null)
dirty= new HashSet<String>();
else
_dirty=null;
return dirty;
}
}
/* ------------------------------------------------------------ */
public Object getVersion()
{
return _version;
}
}

View File

@ -0,0 +1,317 @@
package org.eclipse.jetty.nosql;
//========================================================================
//Copyright (c) 2011 Intalio, Inc.
//------------------------------------------------------------------------
//All rights reserved. This program and the accompanying materials
//are made available under the terms of the Eclipse Public License v1.0
//and Apache License v2.0 which accompanies this distribution.
//The Eclipse Public License is available at
//http://www.eclipse.org/legal/epl-v10.html
//The Apache License v2.0 is available at
//http://www.opensource.org/licenses/apache2.0.php
//You may elect to redistribute this code under either of these licenses.
//========================================================================
import java.util.ArrayList;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import javax.servlet.http.HttpServletRequest;
import org.eclipse.jetty.server.SessionManager;
import org.eclipse.jetty.server.session.AbstractSession;
import org.eclipse.jetty.server.session.AbstractSessionManager;
import org.eclipse.jetty.util.log.Log;
import org.eclipse.jetty.util.log.Logger;
public abstract class NoSqlSessionManager extends AbstractSessionManager implements SessionManager
{
private final static Logger __log = Log.getLogger("org.eclipse.jetty.server.session");
protected final ConcurrentMap<String,NoSqlSession> _sessions=new ConcurrentHashMap<String,NoSqlSession>();
private int _stalePeriod=0;
private int _savePeriod=0;
private int _idlePeriod=-1;
private boolean _invalidateOnStop;
private boolean _saveAllAttributes;
/* ------------------------------------------------------------ */
/* (non-Javadoc)
* @see org.eclipse.jetty.server.session.AbstractSessionManager#doStart()
*/
@Override
public void doStart() throws Exception
{
super.doStart();
}
/* ------------------------------------------------------------ */
@Override
protected void addSession(AbstractSession session)
{
if (isRunning())
_sessions.put(session.getClusterId(),(NoSqlSession)session);
}
/* ------------------------------------------------------------ */
@Override
public AbstractSession getSession(String idInCluster)
{
NoSqlSession session = _sessions.get(idInCluster);
__log.debug("getSession: " + session );
if (session==null)
{
session=loadSession(idInCluster);
if (session!=null)
{
NoSqlSession race=_sessions.putIfAbsent(idInCluster,session);
if (race!=null)
{
session.willPassivate();
session.clearAttributes();
session=race;
}
}
}
return session;
}
/* ------------------------------------------------------------ */
@Override
protected void invalidateSessions() throws Exception
{
// Invalidate all sessions to cause unbind events
ArrayList<NoSqlSession> sessions=new ArrayList<NoSqlSession>(_sessions.values());
int loop=100;
while (sessions.size()>0 && loop-->0)
{
// If we are called from doStop
if (isStopping())
{
// Then we only save and remove the session - it is not invalidated.
for (NoSqlSession session : sessions)
{
session.save(false);
removeSession(session,false);
}
}
else
{
for (NoSqlSession session : sessions)
session.invalidate();
}
// check that no new sessions were created while we were iterating
sessions=new ArrayList<NoSqlSession>(_sessions.values());
}
}
/* ------------------------------------------------------------ */
@Override
protected AbstractSession newSession(HttpServletRequest request)
{
long created=System.currentTimeMillis();
String clusterId=getSessionIdManager().newSessionId(request,created);
return new NoSqlSession(this,created,created,clusterId);
}
/* ------------------------------------------------------------ */
@Override
protected boolean removeSession(String idInCluster)
{
synchronized (this)
{
NoSqlSession session = _sessions.remove(idInCluster);
try
{
if (session != null)
{
return remove(session);
}
}
catch (Exception e)
{
__log.warn("Problem deleting session id=" + idInCluster,e);
}
return session != null;
}
}
/* ------------------------------------------------------------ */
protected void invalidateSession( String idInCluster )
{
synchronized (this)
{
NoSqlSession session = _sessions.remove(idInCluster);
try
{
if (session != null)
{
remove(session);
}
}
catch (Exception e)
{
__log.warn("Problem deleting session id=" + idInCluster,e);
}
}
/*
* ought we not go to cluster and mark it invalid?
*/
}
/* ------------------------------------------------------------ */
/**
* The State Period is the maximum time in seconds that an in memory session is allows to be stale:
* <ul>
* <li>If this period is exceeded, the DB will be checked to see if a more recent version is available.</li>
* <li>If the state period is set to a value < 0, then no staleness check will be made.</li>
* <li>If the state period is set to 0, then a staleness check is made whenever the active request count goes from 0 to 1.</li>
* </ul>
* @return the stalePeriod in seconds
*/
public int getStalePeriod()
{
return _stalePeriod;
}
/* ------------------------------------------------------------ */
/**
* The State Period is the maximum time in seconds that an in memory session is allows to be stale:
* <ul>
* <li>If this period is exceeded, the DB will be checked to see if a more recent version is available.</li>
* <li>If the state period is set to a value < 0, then no staleness check will be made.</li>
* <li>If the state period is set to 0, then a staleness check is made whenever the active request count goes from 0 to 1.</li>
* </ul>
* @param stalePeriod the stalePeriod in seconds
*/
public void setStalePeriod(int stalePeriod)
{
_stalePeriod = stalePeriod;
}
/* ------------------------------------------------------------ */
/**
* The Save Period is the time in seconds between saves of a dirty session to the DB.
* When this period is exceeded, the a dirty session will be written to the DB: <ul>
* <li>a save period of -2 means the session is written to the DB whenever setAttribute is called.</li>
* <li>a save period of -1 means the session is never saved to the DB other than on a shutdown</li>
* <li>a save period of 0 means the session is written to the DB whenever the active request count goes from 1 to 0.</li>
* <li>a save period of 1 means the session is written to the DB whenever the active request count goes from 1 to 0 and the session is dirty.</li>
* <li>a save period of > 1 means the session is written after that period in seconds of being dirty.</li>
* </ul>
* @return the savePeriod -2,-1,0,1 or the period in seconds >=2
*/
public int getSavePeriod()
{
return _savePeriod;
}
/* ------------------------------------------------------------ */
/**
* The Save Period is the time in seconds between saves of a dirty session to the DB.
* When this period is exceeded, the a dirty session will be written to the DB: <ul>
* <li>a save period of -2 means the session is written to the DB whenever setAttribute is called.</li>
* <li>a save period of -1 means the session is never saved to the DB other than on a shutdown</li>
* <li>a save period of 0 means the session is written to the DB whenever the active request count goes from 1 to 0.</li>
* <li>a save period of 1 means the session is written to the DB whenever the active request count goes from 1 to 0 and the session is dirty.</li>
* <li>a save period of > 1 means the session is written after that period in seconds of being dirty.</li>
* </ul>
* @param savePeriod the savePeriod -2,-1,0,1 or the period in seconds >=2
*/
public void setSavePeriod(int savePeriod)
{
_savePeriod = savePeriod;
}
/* ------------------------------------------------------------ */
/**
* The Idle Period is the time in seconds before an in memory session is passivated.
* When this period is exceeded, the session will be passivated and removed from memory. If the session was dirty, it will be written to the DB.
* If the idle period is set to a value < 0, then the session is never idled.
* If the save period is set to 0, then the session is idled whenever the active request count goes from 1 to 0.
* @return the idlePeriod
*/
public int getIdlePeriod()
{
return _idlePeriod;
}
/* ------------------------------------------------------------ */
/**
* The Idle Period is the time in seconds before an in memory session is passivated.
* When this period is exceeded, the session will be passivated and removed from memory. If the session was dirty, it will be written to the DB.
* If the idle period is set to a value < 0, then the session is never idled.
* If the save period is set to 0, then the session is idled whenever the active request count goes from 1 to 0.
* @param idlePeriod the idlePeriod in seconds
*/
public void setIdlePeriod(int idlePeriod)
{
_idlePeriod = idlePeriod;
}
/* ------------------------------------------------------------ */
/**
* Invalidate sessions when the session manager is stopped otherwise save them to the DB.
* @return the invalidateOnStop
*/
public boolean isInvalidateOnStop()
{
return _invalidateOnStop;
}
/* ------------------------------------------------------------ */
/**
* Invalidate sessions when the session manager is stopped otherwise save them to the DB.
* @param invalidateOnStop the invalidateOnStop to set
*/
public void setInvalidateOnStop(boolean invalidateOnStop)
{
_invalidateOnStop = invalidateOnStop;
}
/* ------------------------------------------------------------ */
/**
* Save all attributes of a session or only update the dirty attributes.
* @return the saveAllAttributes
*/
public boolean isSaveAllAttributes()
{
return _saveAllAttributes;
}
/* ------------------------------------------------------------ */
/**
* Save all attributes of a session or only update the dirty attributes.
* @param saveAllAttributes the saveAllAttributes to set
*/
public void setSaveAllAttributes(boolean saveAllAttributes)
{
_saveAllAttributes = saveAllAttributes;
}
/* ------------------------------------------------------------ */
abstract protected NoSqlSession loadSession(String clusterId);
/* ------------------------------------------------------------ */
abstract protected Object save(NoSqlSession session,Object version, boolean activateAfterSave);
/* ------------------------------------------------------------ */
abstract protected Object refresh(NoSqlSession session, Object version);
/* ------------------------------------------------------------ */
abstract protected boolean remove(NoSqlSession session);
}

View File

@ -0,0 +1,540 @@
package org.eclipse.jetty.nosql.mongodb;
//========================================================================
//Copyright (c) 2011 Intalio, Inc.
//------------------------------------------------------------------------
//All rights reserved. This program and the accompanying materials
//are made available under the terms of the Eclipse Public License v1.0
//and Apache License v2.0 which accompanies this distribution.
//The Eclipse Public License is available at
//http://www.eclipse.org/legal/epl-v10.html
//The Apache License v2.0 is available at
//http://www.opensource.org/licenses/apache2.0.php
//You may elect to redistribute this code under either of these licenses.
//========================================================================
import java.net.UnknownHostException;
import java.util.HashSet;
import java.util.Random;
import java.util.Set;
import java.util.Timer;
import java.util.TimerTask;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpSession;
import org.eclipse.jetty.server.Handler;
import org.eclipse.jetty.server.Server;
import org.eclipse.jetty.server.SessionManager;
import org.eclipse.jetty.server.handler.ContextHandler;
import org.eclipse.jetty.server.session.AbstractSessionIdManager;
import org.eclipse.jetty.server.session.SessionHandler;
import org.eclipse.jetty.util.log.Log;
import org.eclipse.jetty.util.log.Logger;
import com.mongodb.BasicDBObject;
import com.mongodb.BasicDBObjectBuilder;
import com.mongodb.DBCollection;
import com.mongodb.DBCursor;
import com.mongodb.DBObject;
import com.mongodb.Mongo;
import com.mongodb.MongoException;
/**
* Based partially on the jdbc session id manager...
*
* Theory is that we really only need the session id manager for the local
* instance so we have something to scavenge on, namely the list of known ids
*
* this class has a timer that runs at the scavenge delay that runs a query
* for all id's known to this node and that have and old accessed value greater
* then the scavengeDelay.
*
* these found sessions are then run through the invalidateAll(id) method that
* is a bit hinky but is supposed to notify all handlers this id is now DOA and
* ought to be cleaned up. this ought to result in a save operation on the session
* that will change the valid field to false (this conjecture is unvalidated atm)
*/
public class MongoSessionIdManager extends AbstractSessionIdManager
{
private final static Logger __log = Log.getLogger("org.eclipse.jetty.server.session");
final static DBObject __version_1 = new BasicDBObject(MongoSessionManager.__VERSION,1);
final static DBObject __valid_false = new BasicDBObject(MongoSessionManager.__VALID,false);
final DBCollection _sessions;
protected Server _server;
private Timer _scavengeTimer;
private Timer _purgeTimer;
private TimerTask _scavengerTask;
private TimerTask _purgeTask;
private long _scavengeDelay = 30 * 60 * 1000; // every 30 minutes
private long _scavengePeriod = 10 * 6 * 1000; // wait at least 10 minutes
/**
* purge process is enabled by default
*/
private boolean _purge = true;
/**
* purge process would run daily by default
*/
private long _purgeDelay = 24 * 60 * 60 * 1000; // every day
/**
* how long do you want to persist sessions that are no longer
* valid before removing them completely
*/
private long _purgeInvalidAge = 24 * 60 * 60 * 1000; // default 1 day
/**
* how long do you want to leave sessions that are still valid before
* assuming they are dead and removing them
*/
private long _purgeValidAge = 7 * 24 * 60 * 60 * 1000; // default 1 week
/**
* the collection of session ids known to this manager
*
* TODO consider if this ought to be concurrent or not
*/
protected final Set<String> _sessionsIds = new HashSet<String>();
/* ------------------------------------------------------------ */
public MongoSessionIdManager(Server server) throws UnknownHostException, MongoException
{
this(server, new Mongo().getDB("HttpSessions").getCollection("sessions"));
}
/* ------------------------------------------------------------ */
public MongoSessionIdManager(Server server, DBCollection sessions)
{
super(new Random());
_server = server;
_sessions = sessions;
_sessions.ensureIndex(
BasicDBObjectBuilder.start().add("id",1).get(),
BasicDBObjectBuilder.start().add("unique",true).add("sparse",false).get());
_sessions.ensureIndex(
BasicDBObjectBuilder.start().add("id",1).add("version",1).get(),
BasicDBObjectBuilder.start().add("unique",true).add("sparse",false).get());
}
/* ------------------------------------------------------------ */
/**
* Scavenge is a process that periodically checks the tracked session
* ids of this given instance of the session id manager to see if they
* are past the point of expiration.
*/
protected void scavenge()
{
__log.debug("SessionIdManager:scavenge:called with delay" + _scavengeDelay);
synchronized (_sessionsIds)
{
/*
* run a query returning results that:
* - are in the known list of sessionIds
* - have an accessed time less then current time - the scavenger period
*
* we limit the query to return just the __ID so we are not sucking back full sessions
*/
BasicDBObject query = new BasicDBObject();
query.put(MongoSessionManager.__ID,new BasicDBObject("$in", _sessionsIds ));
query.put(MongoSessionManager.__ACCESSED, new BasicDBObject("$lt",System.currentTimeMillis() - _scavengeDelay));
DBCursor checkSessions = _sessions.find(query, new BasicDBObject(MongoSessionManager.__ID, 1));
for ( DBObject session : checkSessions )
{
invalidateAll((String)session.get(MongoSessionManager.__ID));
}
}
}
/* ------------------------------------------------------------ */
/**
* ScavengeFully is a process that periodically checks the tracked session
* ids of this given instance of the session id manager to see if they
* are past the point of expiration.
*
* NOTE: this is potentially devastating and may lead to serious session
* coherence issues, not to be used in a running cluster
*/
protected void scavengeFully()
{
__log.debug("SessionIdManager:scavengeFully");
DBCursor checkSessions = _sessions.find();
for (DBObject session : checkSessions)
{
invalidateAll((String)session.get(MongoSessionManager.__ID));
}
}
/* ------------------------------------------------------------ */
/**
* Purge is a process that cleans the mongodb cluster of old sessions that are no
* longer valid.
*
* There are two checks being done here:
*
* - if the accessed time is older then the current time minus the purge invalid age
* and it is no longer valid then remove that session
* - if the accessed time is older then the current time minus the purge valid age
* then we consider this a lost record and remove it
*
* NOTE: if your system supports long lived sessions then the purge valid age should be
* set to zero so the check is skipped.
*
* The second check was added to catch sessions that were being managed on machines
* that might have crashed without marking their sessions as 'valid=false'
*/
protected void purge()
{
BasicDBObject invalidQuery = new BasicDBObject();
invalidQuery.put(MongoSessionManager.__ACCESSED, new BasicDBObject("$lt",System.currentTimeMillis() - _purgeInvalidAge));
invalidQuery.put(MongoSessionManager.__VALID, __valid_false);
DBCursor oldSessions = _sessions.find(invalidQuery, new BasicDBObject(MongoSessionManager.__ID, 1));
for (DBObject session : oldSessions)
{
String id = (String)session.get("id");
__log.debug("MongoSessionIdManager:purging invalid " + id);
_sessions.remove(session);
}
if (_purgeValidAge != 0)
{
BasicDBObject validQuery = new BasicDBObject();
validQuery.put(MongoSessionManager.__ACCESSED,new BasicDBObject("$lt",System.currentTimeMillis() - _purgeValidAge));
validQuery.put(MongoSessionManager.__VALID, __valid_false);
oldSessions = _sessions.find(invalidQuery,new BasicDBObject(MongoSessionManager.__ID,1));
for (DBObject session : oldSessions)
{
String id = (String)session.get(MongoSessionManager.__ID);
__log.debug("MongoSessionIdManager:purging valid " + id);
_sessions.remove(session);
}
}
}
/* ------------------------------------------------------------ */
/**
* Purge is a process that cleans the mongodb cluster of old sessions that are no
* longer valid.
*
*/
protected void purgeFully()
{
BasicDBObject invalidQuery = new BasicDBObject();
invalidQuery.put(MongoSessionManager.__VALID, false);
DBCursor oldSessions = _sessions.find(invalidQuery, new BasicDBObject(MongoSessionManager.__ID, 1));
for (DBObject session : oldSessions)
{
String id = (String)session.get(MongoSessionManager.__ID);
__log.debug("MongoSessionIdManager:purging invalid " + id);
_sessions.remove(session);
}
}
/* ------------------------------------------------------------ */
public DBCollection getSessions()
{
return _sessions;
}
/* ------------------------------------------------------------ */
public boolean isPurgeEnabled()
{
return _purge;
}
/* ------------------------------------------------------------ */
public void setPurge(boolean purge)
{
this._purge = purge;
}
/* ------------------------------------------------------------ */
/**
* sets the scavengeDelay
*/
public void setScavengeDelay(long scavengeDelay)
{
this._scavengeDelay = scavengeDelay;
}
/* ------------------------------------------------------------ */
public void setScavengePeriod(long scavengePeriod)
{
this._scavengePeriod = scavengePeriod;
}
/* ------------------------------------------------------------ */
public void setPurgeDelay(long purgeDelay)
{
if ( isRunning() )
{
throw new IllegalStateException();
}
this._purgeDelay = purgeDelay;
}
/* ------------------------------------------------------------ */
public long getPurgeInvalidAge()
{
return _purgeInvalidAge;
}
/* ------------------------------------------------------------ */
/**
* sets how old a session is to be persisted past the point it is
* no longer valid
*/
public void setPurgeInvalidAge(long purgeValidAge)
{
this._purgeInvalidAge = purgeValidAge;
}
/* ------------------------------------------------------------ */
public long getPurgeValidAge()
{
return _purgeValidAge;
}
/* ------------------------------------------------------------ */
/**
* sets how old a session is to be persist past the point it is
* considered no longer viable and should be removed
*
* NOTE: set this value to 0 to disable purging of valid sessions
*/
public void setPurgeValidAge(long purgeValidAge)
{
this._purgeValidAge = purgeValidAge;
}
/* ------------------------------------------------------------ */
@Override
protected void doStart() throws Exception
{
__log.debug("MongoSessionIdManager:starting");
/*
* setup the scavenger thread
*/
if (_scavengeDelay > 0)
{
_scavengeTimer = new Timer("MongoSessionIdScavenger",true);
synchronized (this)
{
if (_scavengerTask != null)
{
_scavengerTask.cancel();
}
_scavengerTask = new TimerTask()
{
@Override
public void run()
{
scavenge();
}
};
_scavengeTimer.schedule(_scavengerTask,_scavengeDelay,_scavengePeriod);
}
}
/*
* if purging is enabled, setup the purge thread
*/
if ( _purge )
{
_purgeTimer = new Timer("MongoSessionPurger", true);
synchronized (this)
{
if (_purgeTask != null)
{
_purgeTask.cancel();
}
_purgeTask = new TimerTask()
{
@Override
public void run()
{
purge();
}
};
_purgeTimer.schedule(_purgeTask,_purgeDelay);
}
}
}
/* ------------------------------------------------------------ */
@Override
protected void doStop() throws Exception
{
if (_scavengeTimer != null)
{
_scavengeTimer.cancel();
_scavengeTimer = null;
}
if (_purgeTimer != null)
{
_purgeTimer.cancel();
_purgeTimer = null;
}
super.doStop();
}
/* ------------------------------------------------------------ */
/**
* is the session id known to mongo, and is it valid
*/
@Override
public boolean idInUse(String sessionId)
{
/*
* optimize this query to only return the valid variable
*/
DBObject o = _sessions.findOne(new BasicDBObject("id",sessionId), __valid_false);
if ( o != null )
{
Boolean valid = (Boolean)o.get(MongoSessionManager.__VALID);
if ( valid == null )
{
return false;
}
return valid;
}
return false;
}
/* ------------------------------------------------------------ */
@Override
public void addSession(HttpSession session)
{
if (session == null)
{
return;
}
/*
* already a part of the index in mongo...
*/
__log.debug("MongoSessionIdManager:addSession:" + session.getId());
synchronized (_sessionsIds)
{
_sessionsIds.add(session.getId());
}
}
/* ------------------------------------------------------------ */
@Override
public void removeSession(HttpSession session)
{
if (session == null)
{
return;
}
synchronized (_sessionsIds)
{
_sessionsIds.remove(session.getId());
}
}
/* ------------------------------------------------------------ */
@Override
public void invalidateAll(String sessionId)
{
synchronized (_sessionsIds)
{
_sessionsIds.remove(sessionId);
//tell all contexts that may have a session object with this id to
//get rid of them
Handler[] contexts = _server.getChildHandlersByClass(ContextHandler.class);
for (int i=0; contexts!=null && i<contexts.length; i++)
{
SessionHandler sessionHandler = (SessionHandler)((ContextHandler)contexts[i]).getChildHandlerByClass(SessionHandler.class);
if (sessionHandler != null)
{
SessionManager manager = sessionHandler.getSessionManager();
if (manager != null && manager instanceof MongoSessionManager)
{
((MongoSessionManager)manager).invalidateSession(sessionId);
}
}
}
}
}
/* ------------------------------------------------------------ */
// TODO not sure if this is correct
@Override
public String getClusterId(String nodeId)
{
int dot=nodeId.lastIndexOf('.');
return (dot>0)?nodeId.substring(0,dot):nodeId;
}
/* ------------------------------------------------------------ */
// TODO not sure if this is correct
@Override
public String getNodeId(String clusterId, HttpServletRequest request)
{
if (_workerName!=null)
return clusterId+'.'+_workerName;
return clusterId;
}
}

View File

@ -0,0 +1,549 @@
package org.eclipse.jetty.nosql.mongodb;
//========================================================================
//Copyright (c) 2011 Intalio, Inc.
//------------------------------------------------------------------------
//All rights reserved. This program and the accompanying materials
//are made available under the terms of the Eclipse Public License v1.0
//and Apache License v2.0 which accompanies this distribution.
//The Eclipse Public License is available at
//http://www.eclipse.org/legal/epl-v10.html
//The Apache License v2.0 is available at
//http://www.opensource.org/licenses/apache2.0.php
//You may elect to redistribute this code under either of these licenses.
//========================================================================
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.UnknownHostException;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import org.eclipse.jetty.nosql.NoSqlSession;
import org.eclipse.jetty.nosql.NoSqlSessionManager;
import org.eclipse.jetty.server.SessionIdManager;
import org.eclipse.jetty.util.log.Log;
import org.eclipse.jetty.util.log.Logger;
import org.omg.CORBA._IDLTypeStub;
import com.mongodb.BasicDBObject;
import com.mongodb.DBCollection;
import com.mongodb.DBObject;
import com.mongodb.MongoException;
public class MongoSessionManager extends NoSqlSessionManager
{
private final static Logger __log = Log.getLogger("org.eclipse.jetty.server.session");
/*
* strings used as keys or parts of keys in mongo
*/
private final static String __METADATA = "__metadata__";
public final static String __ID = "id";
private final static String __CREATED = "created";
public final static String __VALID = "valid";
public final static String __INVALIDATED = "invalidated";
public final static String __ACCESSED = "accessed";
private final static String __CONTEXT = "context";
public final static String __VERSION = __METADATA + ".version";
/**
* the context id is only set when this class has been started
*/
private String _contextId = null;
private DBCollection _sessions;
private DBObject __version_1;
/* ------------------------------------------------------------ */
public MongoSessionManager() throws UnknownHostException, MongoException
{
}
/*------------------------------------------------------------ */
@Override
public void doStart() throws Exception
{
super.doStart();
String[] hosts = getContextHandler().getVirtualHosts();
if (hosts == null || hosts.length == 0)
hosts = getContextHandler().getConnectorNames();
if (hosts == null || hosts.length == 0)
hosts = new String[]
{ "::" }; // IPv6 equiv of 0.0.0.0
String contextPath = getContext().getContextPath();
if (contextPath == null || "".equals(contextPath))
{
contextPath = "*";
}
_contextId = createContextId(hosts,contextPath);
__version_1 = new BasicDBObject(getContextKey(__VERSION),1);
}
/* ------------------------------------------------------------ */
/* (non-Javadoc)
* @see org.eclipse.jetty.server.session.AbstractSessionManager#setSessionIdManager(org.eclipse.jetty.server.SessionIdManager)
*/
@Override
public void setSessionIdManager(SessionIdManager metaManager)
{
MongoSessionIdManager msim = (MongoSessionIdManager)metaManager;
_sessions=msim.getSessions();
super.setSessionIdManager(metaManager);
}
/* ------------------------------------------------------------ */
@Override
protected synchronized Object save(NoSqlSession session, Object version, boolean activateAfterSave)
{
try
{
__log.debug("MongoSessionManager:save:" + session);
session.willPassivate();
ByteArrayOutputStream bout = new ByteArrayOutputStream();
ObjectOutputStream out = new ObjectOutputStream(bout);
// Form query for upsert
BasicDBObject key = new BasicDBObject(__ID,session.getClusterId());
key.put(__VALID,true);
// Form updates
BasicDBObject update = new BasicDBObject();
boolean upsert = false;
BasicDBObject sets = new BasicDBObject();
BasicDBObject unsets = new BasicDBObject();
// handle new or existing
if (version == null)
{
// New session
upsert = true;
version = new Long(1);
sets.put(__CREATED,session.getCreationTime());
sets.put(getContextKey(__VERSION),version);
}
else
{
version = new Long(((Long)version).intValue() + 1);
update.put("$inc",__version_1);
}
// handle valid or invalid
if (session.isValid())
{
sets.put(__ACCESSED,session.getAccessed());
Set<String> names = session.takeDirty();
if (isSaveAllAttributes() || upsert)
{
names.addAll(session.getNames()); // note dirty may include removed names
}
for (String name : names)
{
Object value = session.getAttribute(name);
if (value == null)
unsets.put(getContextKey() + "." + encodeName(name),1);
else
sets.put(getContextKey() + "." + encodeName(name),encodeName(out,bout,value));
}
}
else
{
sets.put(__VALID,false);
sets.put(__INVALIDATED, System.currentTimeMillis());
unsets.put(getContextKey(),1);
}
// Do the upsert
if (!sets.isEmpty())
update.put("$set",sets);
if (!unsets.isEmpty())
update.put("$unset",unsets);
_sessions.update(key,update,upsert,false);
__log.debug("MongoSessionManager:save:db.sessions.update(" + key + "," + update + ",true)");
if (activateAfterSave)
session.didActivate();
return version;
}
catch (Exception e)
{
Log.warn(e);
}
return null;
}
/*------------------------------------------------------------ */
@Override
protected Object refresh(NoSqlSession session, Object version)
{
__log.debug("MongoSessionManager:refresh " + session);
// check if our in memory version is the same as what is on the disk
if (version != null)
{
DBObject o = _sessions.findOne(new BasicDBObject(__ID,session.getClusterId()),__version_1);
if (o != null)
{
Object saved = getNestedValue(o, getContextKey(__VERSION));
if (saved != null && saved.equals(version))
{
__log.debug("MongoSessionManager:refresh not needed");
return version;
}
version = saved;
}
}
// If we are here, we have to load the object
DBObject o = _sessions.findOne(new BasicDBObject(__ID,session.getClusterId()));
// If it doesn't exist, invalidate
if (o == null)
{
__log.debug("MongoSessionManager:refresh:marking invalid, no object");
session.invalidate();
return null;
}
// If it has been flagged invalid, invalidate
Boolean valid = (Boolean)o.get(__VALID);
if (valid == null || !valid)
{
__log.debug("MongoSessionManager:refresh:marking invalid, valid flag " + valid);
session.invalidate();
return null;
}
// We need to update the attributes. We will model this as a passivate,
// followed by bindings and then activation.
session.willPassivate();
try
{
session.clearAttributes();
DBObject attrs = (DBObject)getNestedValue(o,getContextKey());
if (attrs != null)
{
for (String name : attrs.keySet())
{
if ( __METADATA.equals(name) )
{
continue;
}
String attr = decodeName(name);
Object value = decodeValue(attrs.get(name));
session.doPutOrRemove(attr,value);
session.bindValue(attr,value);
}
}
session.didActivate();
return version;
}
catch (Exception e)
{
Log.warn(e);
}
return null;
}
/*------------------------------------------------------------ */
@Override
protected synchronized NoSqlSession loadSession(String clusterId)
{
DBObject o = _sessions.findOne(new BasicDBObject(__ID,clusterId));
__log.debug("MongoSessionManager:loaded " + o);
if (o == null)
{
return null;
}
Boolean valid = (Boolean)o.get(__VALID);
if (valid == null || !valid)
{
return null;
}
try
{
Object version = o.get(getContextKey(__VERSION));
Long created = (Long)o.get(__CREATED);
Long accessed = (Long)o.get(__ACCESSED);
NoSqlSession session = new NoSqlSession(this,created,accessed,clusterId,version);
// get the attributes for the context
DBObject attrs = (DBObject)getNestedValue(o,getContextKey());
__log.debug("MongoSessionManager:attrs: " + attrs);
if (attrs != null)
{
for (String name : attrs.keySet())
{
if ( __METADATA.equals(name) )
{
continue;
}
String attr = decodeName(name);
Object value = decodeValue(attrs.get(name));
session.doPutOrRemove(attr,value);
session.bindValue(attr,value);
}
}
session.didActivate();
return session;
}
catch (Exception e)
{
Log.warn(e);
}
return null;
}
/*------------------------------------------------------------ */
@Override
protected boolean remove(NoSqlSession session)
{
__log.debug("MongoSessionManager:remove:session " + session.getClusterId());
/*
* Check if the session exists and if it does remove the context
* associated with this session
*/
BasicDBObject key = new BasicDBObject(__ID,session.getClusterId());
DBObject o = _sessions.findOne(key,__version_1);
if (o != null)
{
BasicDBObject remove = new BasicDBObject();
BasicDBObject unsets = new BasicDBObject();
unsets.put(getContextKey(),1);
remove.put("$unsets",unsets);
_sessions.update(key,remove);
return true;
}
else
{
return false;
}
}
/*------------------------------------------------------------ */
@Override
protected void invalidateSession(String idInCluster)
{
__log.debug("MongoSessionManager:invalidateSession:invalidating " + idInCluster);
super.invalidateSession(idInCluster);
/*
* pull back the 'valid' value, we can check if its false, if is we don't need to
* reset it to false
*/
DBObject validKey = new BasicDBObject(__VALID, true);
DBObject o = _sessions.findOne(new BasicDBObject(__ID,idInCluster), validKey);
if (o != null && (Boolean)o.get(__VALID))
{
BasicDBObject update = new BasicDBObject();
BasicDBObject sets = new BasicDBObject();
sets.put(__VALID,false);
sets.put(__INVALIDATED, System.currentTimeMillis());
update.put("$set",sets);
BasicDBObject key = new BasicDBObject(__ID,idInCluster);
_sessions.update(key,update);
}
}
/*------------------------------------------------------------ */
protected String encodeName(String name)
{
return name.replace("%","%25").replace(".","%2E");
}
/*------------------------------------------------------------ */
protected String decodeName(String name)
{
return name.replace("%2E",".").replace("%25","%");
}
/*------------------------------------------------------------ */
protected Object encodeName(ObjectOutputStream out, ByteArrayOutputStream bout, Object value) throws IOException
{
if (value instanceof Number || value instanceof String || value instanceof Boolean || value instanceof Date)
{
return value;
}
else if (value.getClass().equals(HashMap.class))
{
BasicDBObject o = new BasicDBObject();
for (Map.Entry<?, ?> entry : ((Map<?, ?>)value).entrySet())
{
if (!(entry.getKey() instanceof String))
{
o = null;
break;
}
o.append(encodeName(entry.getKey().toString()),encodeName(out,bout,value));
}
if (o != null)
return o;
}
bout.reset();
out.reset();
out.writeUnshared(value);
out.flush();
return bout.toByteArray();
}
/*------------------------------------------------------------ */
protected Object decodeValue(Object value) throws IOException, ClassNotFoundException
{
if (value == null || value instanceof Number || value instanceof String || value instanceof Boolean || value instanceof Date)
{
return value;
}
else if (value instanceof byte[])
{
ObjectInputStream in = new ObjectInputStream(new ByteArrayInputStream((byte[])value));
return in.readObject();
}
else if (value instanceof DBObject)
{
Map<String, Object> map = new HashMap<String, Object>();
for (String name : ((DBObject)value).keySet())
{
String attr = decodeName(name);
map.put(attr,decodeValue(((DBObject)value).get(name)));
}
return map;
}
else
{
throw new IllegalStateException(value.getClass().toString());
}
}
/*------------------------------------------------------------ */
private String getContextKey()
{
return __CONTEXT + "." + _contextId;
}
/*------------------------------------------------------------ */
private String getContextKey(String keybit)
{
return __CONTEXT + "." + _contextId + "." + keybit;
}
public void purge()
{
((MongoSessionIdManager)_sessionIdManager).purge();
}
public void purgeFully()
{
((MongoSessionIdManager)_sessionIdManager).purgeFully();
}
public void scavenge()
{
((MongoSessionIdManager)_sessionIdManager).scavenge();
}
public void scavengeFully()
{
((MongoSessionIdManager)_sessionIdManager).scavengeFully();
}
/*------------------------------------------------------------ */
/**
* returns the total number of session objects in the session store
*
* the count() operation itself is optimized to perform on the server side
* and avoid loading to client side.
*/
public long getSessionStoreCount()
{
return _sessions.find().count();
}
/*------------------------------------------------------------ */
/**
* MongoDB keys are . delimited for nesting so .'s are protected characters
*
* @param virtualHosts
* @param contextPath
* @return
*/
private String createContextId(String[] virtualHosts, String contextPath)
{
String contextId = virtualHosts[0] + contextPath;
contextId.replace('/', '_');
contextId.replace('.','_');
contextId.replace('\\','_');
return contextId;
}
/**
* Dig through a given dbObject for the nested value
*/
private Object getNestedValue(DBObject dbObject, String nestedKey)
{
String[] keyChain = nestedKey.split("\\.");
DBObject temp = dbObject;
for (int i = 0; i < keyChain.length - 1; ++i)
{
temp = (DBObject)temp.get(keyChain[i]);
if ( temp == null )
{
return null;
}
}
return temp.get(keyChain[keyChain.length - 1]);
}
}

View File

@ -0,0 +1,45 @@
package org.eclipse.jetty.nosql.mongodb.jmx;
import org.eclipse.jetty.nosql.mongodb.MongoSessionManager;
import org.eclipse.jetty.server.handler.AbstractHandlerContainer;
import org.eclipse.jetty.server.handler.ContextHandler;
import org.eclipse.jetty.server.session.AbstractSessionManager;
import org.eclipse.jetty.server.session.SessionHandler;
import org.eclipse.jetty.server.session.jmx.AbstractSessionManagerMBean;
public class MongoSessionManagerMBean extends AbstractSessionManagerMBean
{
public MongoSessionManagerMBean(Object managedObject)
{
super(managedObject);
}
/* ------------------------------------------------------------ */
public String getObjectContextBasis()
{
if (_managed != null && _managed instanceof MongoSessionManager)
{
MongoSessionManager manager = (MongoSessionManager)_managed;
String basis = null;
SessionHandler handler = manager.getSessionHandler();
if (handler != null)
{
ContextHandler context =
AbstractHandlerContainer.findContainerOf(handler.getServer(),
ContextHandler.class,
handler);
if (context != null)
basis = getContextName(context);
}
if (basis != null)
return basis;
}
return super.getObjectContextBasis();
}
}

View File

@ -0,0 +1,6 @@
MongoSessionManager: Mongo Session Manager
sessionStoreCount: total number of known sessions in the store
purge(): force a purge() of invalid sessions in the session store based on normal criteria
purgeFully(): force a full purge of invalid sessions in the session store
scavenge(): force a scavenge() of sessions known to this manager in the session store
scavengeFully(): force a scavenge of all sessions in the session store

View File

@ -0,0 +1,35 @@
package org.eclipse.jetty.nosql.mongodb;
// ========================================================================
// Copyright (c) 1996-2009 Mort Bay Consulting Pty. Ltd.
// ------------------------------------------------------------------------
// All rights reserved. This program and the accompanying materials
// are made available under the terms of the Eclipse Public License v1.0
// and Apache License v2.0 which accompanies this distribution.
// The Eclipse Public License is available at
// http://www.eclipse.org/legal/epl-v10.html
// The Apache License v2.0 is available at
// http://www.opensource.org/licenses/apache2.0.php
// You may elect to redistribute this code under either of these licenses.
// ========================================================================
import org.eclipse.jetty.server.session.AbstractClientCrossContextSessionTest;
import org.eclipse.jetty.server.session.AbstractTestServer;
import org.junit.Ignore;
import org.junit.Test;
public class ClientCrossContextSessionTest extends AbstractClientCrossContextSessionTest
{
public AbstractTestServer createServer(int port)
{
return new MongoTestServer(port);
}
@Test
@Ignore ("requires mongodb server")
public void testCrossContextDispatch() throws Exception
{
super.testCrossContextDispatch();
}
}

View File

@ -0,0 +1,34 @@
package org.eclipse.jetty.nosql.mongodb;
//========================================================================
//Copyright (c) 2011 Intalio, Inc.
//------------------------------------------------------------------------
//All rights reserved. This program and the accompanying materials
//are made available under the terms of the Eclipse Public License v1.0
//and Apache License v2.0 which accompanies this distribution.
//The Eclipse Public License is available at
//http://www.eclipse.org/legal/epl-v10.html
//The Apache License v2.0 is available at
//http://www.opensource.org/licenses/apache2.0.php
//You may elect to redistribute this code under either of these licenses.
//========================================================================
import org.eclipse.jetty.server.session.AbstractLastAccessTimeTest;
import org.eclipse.jetty.server.session.AbstractTestServer;
import org.junit.Ignore;
import org.junit.Test;
public class LastAccessTimeTest extends AbstractLastAccessTimeTest
{
public AbstractTestServer createServer(int port, int max, int scavenge)
{
return new MongoTestServer(port,max,scavenge);
}
@Test
@Ignore ("requires mongodb server")
public void testLastAccessTime() throws Exception
{
super.testLastAccessTime();
}
}

View File

@ -0,0 +1,39 @@
package org.eclipse.jetty.nosql.mongodb;
// ========================================================================
// Copyright (c) 1996-2009 Mort Bay Consulting Pty. Ltd.
// ------------------------------------------------------------------------
// All rights reserved. This program and the accompanying materials
// are made available under the terms of the Eclipse Public License v1.0
// and Apache License v2.0 which accompanies this distribution.
// The Eclipse Public License is available at
// http://www.eclipse.org/legal/epl-v10.html
// The Apache License v2.0 is available at
// http://www.opensource.org/licenses/apache2.0.php
// You may elect to redistribute this code under either of these licenses.
// ========================================================================
import org.eclipse.jetty.server.session.AbstractLightLoadTest;
import org.eclipse.jetty.server.session.AbstractTestServer;
import org.junit.Ignore;
import org.junit.Test;
/**
* LightLoadTest
*/
public class LightLoadTest extends AbstractLightLoadTest
{
public AbstractTestServer createServer(int port)
{
return new MongoTestServer(port);
}
@Test
@Ignore ("requires mongodb server")
public void testLightLoad() throws Exception
{
super.testLightLoad();
}
}

View File

@ -0,0 +1,60 @@
package org.eclipse.jetty.nosql.mongodb;
//========================================================================
//Copyright (c) 2011 Intalio, Inc.
//------------------------------------------------------------------------
//All rights reserved. This program and the accompanying materials
//are made available under the terms of the Eclipse Public License v1.0
//and Apache License v2.0 which accompanies this distribution.
//The Eclipse Public License is available at
//http://www.eclipse.org/legal/epl-v10.html
//The Apache License v2.0 is available at
//http://www.opensource.org/licenses/apache2.0.php
//You may elect to redistribute this code under either of these licenses.
//========================================================================
import java.util.Set;
import com.mongodb.BasicDBObject;
import com.mongodb.DB;
import com.mongodb.DBCollection;
import com.mongodb.DBObject;
import com.mongodb.Mongo;
import com.mongodb.WriteResult;
public class MongoTest
{
public static void main(String... args) throws Exception
{
Mongo m = new Mongo( "127.0.0.1" , 27017 );
DB db = m.getDB( "mydb" );
Set<String> colls = db.getCollectionNames();
System.err.println("Colls="+colls);
DBCollection coll = db.getCollection("testCollection");
BasicDBObject key = new BasicDBObject("id","1234");
BasicDBObject sets = new BasicDBObject("name","value");
BasicDBObject upsert=new BasicDBObject("$set",sets);
WriteResult result =coll.update(key,upsert,true,false);
System.err.println(result.getLastError());
while (coll.count()>0)
{
DBObject docZ = coll.findOne();
System.err.println("removing "+ docZ);
if (docZ!=null)
coll.remove(docZ);
}
}
}

View File

@ -0,0 +1,137 @@
package org.eclipse.jetty.nosql.mongodb;
// ========================================================================
// Copyright (c) 1996-2009 Mort Bay Consulting Pty. Ltd.
// ------------------------------------------------------------------------
// All rights reserved. This program and the accompanying materials
// are made available under the terms of the Eclipse Public License v1.0
// and Apache License v2.0 which accompanies this distribution.
// The Eclipse Public License is available at
// http://www.eclipse.org/legal/epl-v10.html
// The Apache License v2.0 is available at
// http://www.opensource.org/licenses/apache2.0.php
// You may elect to redistribute this code under either of these licenses.
// ========================================================================
import java.util.concurrent.TimeUnit;
import org.eclipse.jetty.nosql.mongodb.MongoSessionIdManager;
import org.eclipse.jetty.nosql.mongodb.MongoSessionManager;
import org.eclipse.jetty.server.SessionIdManager;
import org.eclipse.jetty.server.SessionManager;
import org.eclipse.jetty.server.session.AbstractTestServer;
import org.eclipse.jetty.server.session.SessionHandler;
/**
* @version $Revision$ $Date$
*/
public class MongoTestServer extends AbstractTestServer
{
static MongoSessionIdManager _idManager;
private boolean _saveAllAttributes = false; // false save dirty, true save all
public MongoTestServer(int port)
{
super(port, 30, 10);
}
public MongoTestServer(int port, int maxInactivePeriod, int scavengePeriod)
{
super(port, maxInactivePeriod, scavengePeriod);
}
public MongoTestServer(int port, int maxInactivePeriod, int scavengePeriod, boolean saveAllAttributes)
{
super(port, maxInactivePeriod, scavengePeriod);
_saveAllAttributes = saveAllAttributes;
}
public SessionIdManager newSessionIdManager()
{
if ( _idManager != null )
{
try
{
_idManager.stop();
}
catch (Exception e)
{
e.printStackTrace();
}
_idManager.setScavengeDelay(_scavengePeriod + 1000);
_idManager.setScavengePeriod(_maxInactivePeriod);
try
{
_idManager.start();
}
catch (Exception e)
{
e.printStackTrace();
}
return _idManager;
}
try
{
System.err.println("MongoTestServer:SessionIdManager:" + _maxInactivePeriod + "/" + _scavengePeriod);
_idManager = new MongoSessionIdManager(_server);
_idManager.setScavengeDelay((int)TimeUnit.SECONDS.toMillis(_scavengePeriod));
_idManager.setScavengePeriod(_maxInactivePeriod);
return _idManager;
}
catch (Exception e)
{
throw new IllegalStateException();
}
}
public SessionManager newSessionManager()
{
MongoSessionManager manager;
try
{
manager = new MongoSessionManager();
}
catch (Exception e)
{
throw new RuntimeException(e);
}
manager.setSavePeriod(1);
manager.setStalePeriod(0);
manager.setSaveAllAttributes(_saveAllAttributes);
//manager.setScavengePeriod((int)TimeUnit.SECONDS.toMillis(_scavengePeriod));
return manager;
}
public SessionHandler newSessionHandler(SessionManager sessionManager)
{
return new SessionHandler(sessionManager);
}
public static void main(String... args) throws Exception
{
MongoTestServer server8080 = new MongoTestServer(8080);
server8080.addContext("/").addServlet(SessionDump.class,"/");
server8080.start();
MongoTestServer server8081 = new MongoTestServer(8081);
server8081.addContext("/").addServlet(SessionDump.class,"/");
server8081.start();
server8080.join();
server8081.join();
}
}

View File

@ -0,0 +1,38 @@
package org.eclipse.jetty.nosql.mongodb;
// ========================================================================
// Copyright (c) 1996-2009 Mort Bay Consulting Pty. Ltd.
// ------------------------------------------------------------------------
// All rights reserved. This program and the accompanying materials
// are made available under the terms of the Eclipse Public License v1.0
// and Apache License v2.0 which accompanies this distribution.
// The Eclipse Public License is available at
// http://www.eclipse.org/legal/epl-v10.html
// The Apache License v2.0 is available at
// http://www.opensource.org/licenses/apache2.0.php
// You may elect to redistribute this code under either of these licenses.
// ========================================================================
import org.eclipse.jetty.server.session.AbstractNewSessionTest;
import org.eclipse.jetty.server.session.AbstractTestServer;
import org.junit.Ignore;
import org.junit.Test;
/**
* NewSessionTest
*/
public class NewSessionTest extends AbstractNewSessionTest
{
public AbstractTestServer createServer(int port, int max, int scavenge)
{
return new MongoTestServer(port,max,scavenge);
}
@Test
@Ignore ("requires mongodb server")
public void testNewSession() throws Exception
{
super.testNewSession();
}
}

View File

@ -0,0 +1,37 @@
package org.eclipse.jetty.nosql.mongodb;
// ========================================================================
// Copyright (c) 1996-2009 Mort Bay Consulting Pty. Ltd.
// ------------------------------------------------------------------------
// All rights reserved. This program and the accompanying materials
// are made available under the terms of the Eclipse Public License v1.0
// and Apache License v2.0 which accompanies this distribution.
// The Eclipse Public License is available at
// http://www.eclipse.org/legal/epl-v10.html
// The Apache License v2.0 is available at
// http://www.opensource.org/licenses/apache2.0.php
// You may elect to redistribute this code under either of these licenses.
// ========================================================================
import org.eclipse.jetty.server.session.AbstractOrphanedSessionTest;
import org.eclipse.jetty.server.session.AbstractTestServer;
import org.junit.Ignore;
import org.junit.Test;
/**
* OrphanedSessionTest
*/
public class OrphanedSessionTest extends AbstractOrphanedSessionTest
{
public AbstractTestServer createServer(int port, int max, int scavenge)
{
return new MongoTestServer(port,max,scavenge);
}
@Test
@Ignore ("requires mongodb server")
public void testOrphanedSession() throws Exception
{
super.testOrphanedSession();
}
}

View File

@ -0,0 +1,38 @@
package org.eclipse.jetty.nosql.mongodb;
// ========================================================================
// Copyright (c) 1996-2009 Mort Bay Consulting Pty. Ltd.
// ------------------------------------------------------------------------
// All rights reserved. This program and the accompanying materials
// are made available under the terms of the Eclipse Public License v1.0
// and Apache License v2.0 which accompanies this distribution.
// The Eclipse Public License is available at
// http://www.eclipse.org/legal/epl-v10.html
// The Apache License v2.0 is available at
// http://www.opensource.org/licenses/apache2.0.php
// You may elect to redistribute this code under either of these licenses.
// ========================================================================
import org.eclipse.jetty.server.session.AbstractReentrantRequestSessionTest;
import org.eclipse.jetty.server.session.AbstractTestServer;
import org.junit.Ignore;
import org.junit.Test;
/**
* ReentrantRequestSessionTest
*/
public class ReentrantRequestSessionTest extends AbstractReentrantRequestSessionTest
{
public AbstractTestServer createServer(int port)
{
return new MongoTestServer(port);
}
@Test
@Ignore ("requires mongodb server")
public void testReentrantRequestSession() throws Exception
{
super.testReentrantRequestSession();
}
}

View File

@ -0,0 +1,35 @@
package org.eclipse.jetty.nosql.mongodb;
//========================================================================
//Copyright (c) 2011 Intalio, Inc.
//------------------------------------------------------------------------
//All rights reserved. This program and the accompanying materials
//are made available under the terms of the Eclipse Public License v1.0
//and Apache License v2.0 which accompanies this distribution.
//The Eclipse Public License is available at
//http://www.eclipse.org/legal/epl-v10.html
//The Apache License v2.0 is available at
//http://www.opensource.org/licenses/apache2.0.php
//You may elect to redistribute this code under either of these licenses.
//========================================================================
import org.eclipse.jetty.server.session.AbstractRemoveSessionTest;
import org.eclipse.jetty.server.session.AbstractTestServer;
import org.junit.Ignore;
import org.junit.Test;
public class RemoveSessionTest extends AbstractRemoveSessionTest
{
public AbstractTestServer createServer(int port, int max, int scavenge)
{
return new MongoTestServer(port,max,scavenge);
}
@Test
@Ignore ("requires mongodb server")
public void testRemoveSession() throws Exception
{
super.testRemoveSession();
}
}

View File

@ -0,0 +1,34 @@
package org.eclipse.jetty.nosql.mongodb;
// ========================================================================
// Copyright (c) 1996-2009 Mort Bay Consulting Pty. Ltd.
// ------------------------------------------------------------------------
// All rights reserved. This program and the accompanying materials
// are made available under the terms of the Eclipse Public License v1.0
// and Apache License v2.0 which accompanies this distribution.
// The Eclipse Public License is available at
// http://www.eclipse.org/legal/epl-v10.html
// The Apache License v2.0 is available at
// http://www.opensource.org/licenses/apache2.0.php
// You may elect to redistribute this code under either of these licenses.
// ========================================================================
import org.eclipse.jetty.server.session.AbstractServerCrossContextSessionTest;
import org.eclipse.jetty.server.session.AbstractTestServer;
import org.junit.Ignore;
import org.junit.Test;
public class ServerCrossContextSessionTest extends AbstractServerCrossContextSessionTest
{
public AbstractTestServer createServer(int port)
{
return new MongoTestServer(port);
}
@Test
@Ignore ("requires mongodb server")
public void testCrossContextDispatch() throws Exception
{
super.testCrossContextDispatch();
}
}

View File

@ -0,0 +1,181 @@
package org.eclipse.jetty.nosql.mongodb;
// ========================================================================
// Copyright (c) 1996-2009 Mort Bay Consulting Pty. Ltd.
// ------------------------------------------------------------------------
// All rights reserved. This program and the accompanying materials
// are made available under the terms of the Eclipse Public License v1.0
// and Apache License v2.0 which accompanies this distribution.
// The Eclipse Public License is available at
// http://www.eclipse.org/legal/epl-v10.html
// The Apache License v2.0 is available at
// http://www.opensource.org/licenses/apache2.0.php
// You may elect to redistribute this code under either of these licenses.
// ========================================================================
import java.io.IOException;
import java.io.PrintWriter;
import java.util.Date;
import java.util.Enumeration;
import javax.servlet.ServletConfig;
import javax.servlet.ServletException;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import javax.servlet.http.HttpSession;
/* ------------------------------------------------------------ */
/** Test Servlet Sessions.
*
*
*/
public class SessionDump extends HttpServlet
{
int redirectCount=0;
/* ------------------------------------------------------------ */
String pageType;
/* ------------------------------------------------------------ */
@Override
public void init(ServletConfig config)
throws ServletException
{
super.init(config);
}
/* ------------------------------------------------------------ */
protected void handleForm(HttpServletRequest request,
HttpServletResponse response)
{
HttpSession session = request.getSession(false);
String action = request.getParameter("Action");
String name = request.getParameter("Name");
String value = request.getParameter("Value");
if (action!=null)
{
if(action.equals("New Session"))
{
session = request.getSession(true);
session.setAttribute("test","value");
}
else if (session!=null)
{
if (action.equals("Invalidate"))
session.invalidate();
else if (action.equals("Set") && name!=null && name.length()>0)
session.setAttribute(name,value);
else if (action.equals("Remove"))
session.removeAttribute(name);
}
}
}
/* ------------------------------------------------------------ */
@Override
public void doPost(HttpServletRequest request,
HttpServletResponse response)
throws ServletException, IOException
{
handleForm(request,response);
String nextUrl = getURI(request)+"?R="+redirectCount++;
String encodedUrl=response.encodeRedirectURL(nextUrl);
response.sendRedirect(encodedUrl);
}
/* ------------------------------------------------------------ */
@Override
public void doGet(HttpServletRequest request,
HttpServletResponse response)
throws ServletException, IOException
{
handleForm(request,response);
response.setContentType("text/html");
HttpSession session = request.getSession(getURI(request).indexOf("new")>0);
try
{
if (session!=null)
session.isNew();
}
catch(IllegalStateException e)
{
session=null;
}
PrintWriter out = response.getWriter();
out.println("<h1>Session Dump Servlet:</h1>");
out.println("<form action=\""+response.encodeURL(getURI(request))+"\" method=\"post\">");
if (session==null)
{
out.println("<H3>No Session</H3>");
out.println("<input type=\"submit\" name=\"Action\" value=\"New Session\"/>");
}
else
{
try
{
out.println("<b>ID:</b> "+session.getId()+"<br/>");
out.println("<b>New:</b> "+session.isNew()+"<br/>");
out.println("<b>Created:</b> "+new Date(session.getCreationTime())+"<br/>");
out.println("<b>Last:</b> "+new Date(session.getLastAccessedTime())+"<br/>");
out.println("<b>Max Inactive:</b> "+session.getMaxInactiveInterval()+"<br/>");
out.println("<b>Context:</b> "+session.getServletContext()+"<br/>");
Enumeration keys=session.getAttributeNames();
while(keys.hasMoreElements())
{
String name=(String)keys.nextElement();
String value=""+session.getAttribute(name);
out.println("<b>"+name+":</b> "+value+"<br/>");
}
out.println("<b>Name:</b><input type=\"text\" name=\"Name\" /><br/>");
out.println("<b>Value:</b><input type=\"text\" name=\"Value\" /><br/>");
out.println("<input type=\"submit\" name=\"Action\" value=\"Set\"/>");
out.println("<input type=\"submit\" name=\"Action\" value=\"Remove\"/>");
out.println("<input type=\"submit\" name=\"Action\" value=\"Refresh\"/>");
out.println("<input type=\"submit\" name=\"Action\" value=\"Invalidate\"/><br/>");
out.println("</form><br/>");
if (request.isRequestedSessionIdFromCookie())
out.println("<P>Turn off cookies in your browser to try url encoding<BR>");
if (request.isRequestedSessionIdFromURL())
out.println("<P>Turn on cookies in your browser to try cookie encoding<BR>");
out.println("<a href=\""+response.encodeURL(request.getRequestURI()+"?q=0")+"\">Encoded Link</a><BR>");
}
catch (IllegalStateException e)
{
e.printStackTrace();
}
}
}
/* ------------------------------------------------------------ */
@Override
public String getServletInfo() {
return "Session Dump Servlet";
}
/* ------------------------------------------------------------ */
private String getURI(HttpServletRequest request)
{
String uri=(String)request.getAttribute("javax.servlet.forward.request_uri");
if (uri==null)
uri=request.getRequestURI();
return uri;
}
}

View File

@ -0,0 +1,246 @@
package org.eclipse.jetty.nosql.mongodb;
//========================================================================
//Copyright (c) 2011 Intalio, Inc.
//------------------------------------------------------------------------
//All rights reserved. This program and the accompanying materials
//are made available under the terms of the Eclipse Public License v1.0
//and Apache License v2.0 which accompanies this distribution.
//The Eclipse Public License is available at
//http://www.eclipse.org/legal/epl-v10.html
//The Apache License v2.0 is available at
//http://www.opensource.org/licenses/apache2.0.php
//You may elect to redistribute this code under either of these licenses.
//========================================================================
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import java.io.IOException;
import java.io.PrintWriter;
import java.io.Serializable;
import java.lang.management.ManagementFactory;
import java.net.MalformedURLException;
import javax.management.remote.JMXServiceURL;
import javax.servlet.ServletException;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import org.eclipse.jetty.client.ContentExchange;
import org.eclipse.jetty.client.HttpClient;
import org.eclipse.jetty.http.HttpMethods;
import org.eclipse.jetty.jmx.ConnectorServer;
import org.eclipse.jetty.jmx.MBeanContainer;
import org.eclipse.jetty.nosql.NoSqlSession;
import org.eclipse.jetty.server.session.AbstractSessionValueSavingTest;
import org.eclipse.jetty.server.session.AbstractTestServer;
import org.junit.Ignore;
import org.junit.Test;
public class SessionSavingValueTest extends AbstractSessionValueSavingTest
{
public AbstractTestServer createServer(int port, int max, int scavenge)
{
// ConnectorServer srv = null;
try
{
// srv = new ConnectorServer(
// new JMXServiceURL("service:jmx:rmi:///jndi/rmi://localhost:0/jettytest"),
// "org.eclipse.jetty:name=rmiconnectorserver");
// srv.start();
MongoTestServer server = new MongoTestServer(port,max,scavenge,true);
// MBeanContainer mbean = new MBeanContainer(ManagementFactory.getPlatformMBeanServer());
//
// server.getServer().getContainer().addEventListener(mbean);
// server.getServer().addBean(mbean);
//
// mbean.start();
return server;
}
// catch (MalformedURLException e)
// {
// // TODO Auto-generated catch block
// e.printStackTrace();
// }
catch (Exception e)
{
// TODO Auto-generated catch block
e.printStackTrace();
}
return null;
}
@Test
@Ignore ("requires mongodb server")
public void testSessionValueSaving() throws Exception
{
String contextPath = "";
String servletMapping = "/server";
int maxInactivePeriod = 10000;
int scavengePeriod = 20000;
AbstractTestServer server1 = createServer(0,maxInactivePeriod,scavengePeriod);
server1.addContext(contextPath).addServlet(TestServlet.class,servletMapping);
server1.start();
int port1 = server1.getPort();
try
{
HttpClient client = new HttpClient();
client.setConnectorType(HttpClient.CONNECTOR_SOCKET);
client.start();
try
{
String[] sessionTestValue = new String[]
{ "0", "null" };
// Perform one request to server1 to create a session
ContentExchange exchange1 = new ContentExchange(true);
exchange1.setMethod(HttpMethods.GET);
exchange1.setURL("http://localhost:" + port1 + contextPath + servletMapping + "?action=init");
client.send(exchange1);
exchange1.waitForDone();
assertEquals(HttpServletResponse.SC_OK,exchange1.getResponseStatus());
String[] sessionTestResponse = exchange1.getResponseContent().split("/");
assertTrue(Long.parseLong(sessionTestValue[0]) < Long.parseLong(sessionTestResponse[0]));
sessionTestValue = sessionTestResponse;
String sessionCookie = exchange1.getResponseFields().getStringField("Set-Cookie");
assertTrue(sessionCookie != null);
// Mangle the cookie, replacing Path with $Path, etc.
sessionCookie = sessionCookie.replaceFirst("(\\W)(P|p)ath=","$1\\$Path=");
// Perform some request to server2 using the session cookie from the previous request
// This should migrate the session from server1 to server2, and leave server1's
// session in a very stale state, while server2 has a very fresh session.
// We want to test that optimizations done to the saving of the shared lastAccessTime
// do not break the correct working
int requestInterval = 500;
for (int i = 0; i < 10; ++i)
{
ContentExchange exchange2 = new ContentExchange(true);
exchange2.setMethod(HttpMethods.GET);
exchange2.setURL("http://localhost:" + port1 + contextPath + servletMapping);
exchange2.getRequestFields().add("Cookie",sessionCookie);
client.send(exchange2);
exchange2.waitForDone();
assertEquals(HttpServletResponse.SC_OK,exchange2.getResponseStatus());
sessionTestResponse = exchange2.getResponseContent().split("/");
assertTrue(Long.parseLong(sessionTestValue[0]) < Long.parseLong(sessionTestResponse[0]));
assertTrue(Long.parseLong(sessionTestValue[1]) < Long.parseLong(sessionTestResponse[1]));
sessionTestValue = sessionTestResponse;
String setCookie = exchange1.getResponseFields().getStringField("Set-Cookie");
if (setCookie != null)
sessionCookie = setCookie.replaceFirst("(\\W)(P|p)ath=","$1\\$Path=");
Thread.sleep(requestInterval);
}
// Thread.sleep(320000);
}
finally
{
client.stop();
}
}
finally
{
server1.stop();
}
}
public static class TestServlet extends HttpServlet
{
@Override
protected void doGet(HttpServletRequest request, HttpServletResponse httpServletResponse) throws ServletException, IOException
{
String action = request.getParameter("action");
if ("init".equals(action))
{
NoSqlSession session = (NoSqlSession)request.getSession(true);
session.setAttribute("test",System.currentTimeMillis());
session.setAttribute("objectTest", new Pojo("foo","bar"));
sendResult(session,httpServletResponse.getWriter());
}
else
{
NoSqlSession session = (NoSqlSession)request.getSession(false);
if (session != null)
{
long value = System.currentTimeMillis();
session.setAttribute("test",value);
}
sendResult(session,httpServletResponse.getWriter());
Pojo p = (Pojo)session.getAttribute("objectTest");
//System.out.println(p.getName() + " / " + p.getValue() );
}
}
private void sendResult(NoSqlSession session, PrintWriter writer)
{
if (session != null)
{
if (session.getVersion() == null)
{
writer.print(session.getAttribute("test") + "/-1");
}
else
{
writer.print(session.getAttribute("test") + "/" + session.getVersion());
}
}
else
{
writer.print("0/-1");
}
}
public class Pojo implements Serializable
{
private String _name;
private String _value;
public Pojo( String name, String value )
{
_name = name;
_value = value;
}
public String getName()
{
return _name;
}
public String getValue()
{
return _value;
}
}
}
}

View File

@ -35,6 +35,7 @@ import org.eclipse.jetty.security.DefaultIdentityService;
import org.eclipse.jetty.security.IdentityService;
import org.eclipse.jetty.security.LoginService;
import org.eclipse.jetty.server.UserIdentity;
import org.eclipse.jetty.util.Loader;
import org.eclipse.jetty.util.component.AbstractLifeCycle;
import org.eclipse.jetty.util.log.Log;
@ -173,28 +174,38 @@ public class JAASLoginService extends AbstractLifeCycle implements LoginService
{
try
{
CallbackHandler callbackHandler = new CallbackHandler()
CallbackHandler callbackHandler = null;
if (_callbackHandlerClass == null)
{
public void handle(Callback[] callbacks) throws IOException, UnsupportedCallbackException
callbackHandler = new CallbackHandler()
{
for (Callback callback: callbacks)
public void handle(Callback[] callbacks) throws IOException, UnsupportedCallbackException
{
if (callback instanceof NameCallback)
for (Callback callback: callbacks)
{
((NameCallback)callback).setName(username);
}
else if (callback instanceof PasswordCallback)
{
((PasswordCallback)callback).setPassword((char[]) credentials.toString().toCharArray());
}
else if (callback instanceof ObjectCallback)
{
((ObjectCallback)callback).setObject(credentials);
if (callback instanceof NameCallback)
{
((NameCallback)callback).setName(username);
}
else if (callback instanceof PasswordCallback)
{
((PasswordCallback)callback).setPassword((char[]) credentials.toString().toCharArray());
}
else if (callback instanceof ObjectCallback)
{
((ObjectCallback)callback).setObject(credentials);
}
}
}
}
};
};
}
else
{
Class clazz = Loader.loadClass(getClass(), _callbackHandlerClass);
callbackHandler = (CallbackHandler)clazz.newInstance();
}
//set up the login context
//TODO jaspi requires we provide the Configuration parameter
Subject subject = new Subject();
@ -220,6 +231,18 @@ public class JAASLoginService extends AbstractLifeCycle implements LoginService
{
Log.warn(e);
}
catch (InstantiationException e)
{
Log.warn(e);
}
catch (IllegalAccessException e)
{
Log.warn(e);
}
catch (ClassNotFoundException e)
{
Log.warn(e);
}
return null;
}

View File

@ -0,0 +1,120 @@
// ========================================================================
// Copyright (c) 2011 Mort Bay Consulting Pty. Ltd.
// ------------------------------------------------------------------------
// All rights reserved. This program and the accompanying materials
// are made available under the terms of the Eclipse Public License v1.0
// and Apache License v2.0 which accompanies this distribution.
// The Eclipse Public License is available at
// http://www.eclipse.org/legal/epl-v10.html
// The Apache License v2.0 is available at
// http://www.opensource.org/licenses/apache2.0.php
// You may elect to redistribute this code under either of these licenses.
// ========================================================================
package org.eclipse.jetty.servlet.listener;
import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.Iterator;
import java.util.concurrent.ConcurrentHashMap;
import javax.servlet.ServletContextEvent;
import javax.servlet.ServletContextListener;
import org.eclipse.jetty.util.Loader;
import org.eclipse.jetty.util.log.Log;
/**
* ELContextCleaner
*
* Clean up BeanELResolver when the context is going out
* of service:
*
* See http://java.net/jira/browse/GLASSFISH-1649
* See https://bugs.eclipse.org/bugs/show_bug.cgi?id=353095
*/
public class ELContextCleaner implements ServletContextListener
{
public void contextInitialized(ServletContextEvent sce)
{
}
public void contextDestroyed(ServletContextEvent sce)
{
try
{
//Check that the BeanELResolver class is on the classpath
Class beanELResolver = Loader.loadClass(this.getClass(), "javax.el.BeanELResolver");
//Get a reference via reflection to the properties field which is holding class references
Field field = getField(beanELResolver);
//Get rid of references
purgeEntries(field);
Log.info("javax.el.BeanELResolver purged");
}
catch (ClassNotFoundException e)
{
//BeanELResolver not on classpath, ignore
}
catch (SecurityException e)
{
Log.warn("Cannot purge classes from javax.el.BeanELResolver", e);
}
catch (IllegalArgumentException e)
{
Log.warn("Cannot purge classes from javax.el.BeanELResolver", e);
}
catch (IllegalAccessException e)
{
Log.warn("Cannot purge classes from javax.el.BeanELResolver", e);
}
catch (NoSuchFieldException e)
{
Log.warn("Cannot purge classes from javax.el.BeanELResolver", e);
}
}
protected Field getField (Class beanELResolver)
throws SecurityException, NoSuchFieldException
{
if (beanELResolver == null)
return null;
return beanELResolver.getDeclaredField("properties");
}
protected void purgeEntries (Field properties)
throws IllegalArgumentException, IllegalAccessException
{
if (properties == null)
return;
if (!properties.isAccessible())
properties.setAccessible(true);
ConcurrentHashMap map = (ConcurrentHashMap) properties.get(null);
if (map == null)
return;
Iterator<Class> itor = map.keySet().iterator();
while (itor.hasNext())
{
Class clazz = itor.next();
Log.info("Clazz: "+clazz+" loaded by "+clazz.getClassLoader());
if (Thread.currentThread().getContextClassLoader().equals(clazz.getClassLoader()))
{
itor.remove();
Log.info("removed");
}
else
Log.info("not removed: "+"contextclassloader="+Thread.currentThread().getContextClassLoader()+"clazz's classloader="+clazz.getClassLoader());
}
}
}

View File

@ -0,0 +1,40 @@
// ========================================================================
// Copyright (c) 2011 Mort Bay Consulting Pty. Ltd.
// ------------------------------------------------------------------------
// All rights reserved. This program and the accompanying materials
// are made available under the terms of the Eclipse Public License v1.0
// and Apache License v2.0 which accompanies this distribution.
// The Eclipse Public License is available at
// http://www.eclipse.org/legal/epl-v10.html
// The Apache License v2.0 is available at
// http://www.opensource.org/licenses/apache2.0.php
// You may elect to redistribute this code under either of these licenses.
// ========================================================================
package org.eclipse.jetty.servlet.listener;
import javax.servlet.ServletContextEvent;
import javax.servlet.ServletContextListener;
/**
* IntrospectorCleaner
*
* Cleans a static cache of Methods held by java.beans.Introspector
* class when a context is undeployed.
*
* @see java.beans.Introspector
*/
public class IntrospectorCleaner implements ServletContextListener
{
public void contextInitialized(ServletContextEvent sce)
{
}
public void contextDestroyed(ServletContextEvent sce)
{
java.beans.Introspector.flushCaches();
}
}

View File

@ -30,6 +30,22 @@
This file is applied to a Web application before it's own WEB_INF/web.xml file
</description>
<!-- ==================================================================== -->
<!-- Removes static references to beans from javax.el.BeanELResolver to -->
<!-- ensure webapp classloader can be released on undeploy -->
<!-- ==================================================================== -->
<listener>
<listener-class>org.eclipse.jetty.servlet.listener.ELContextCleaner</listener-class>
</listener>
<!-- ==================================================================== -->
<!-- Removes static cache of Methods from java.beans.Introspector to -->
<!-- ensure webapp classloader can be released on undeploy -->
<!-- ==================================================================== -->
<listener>
<listener-class>org.eclipse.jetty.servlet.listener.IntrospectorCleaner</listener-class>
</listener>
<!-- ==================================================================== -->
<!-- Context params to control Session Cookies -->

View File

@ -264,7 +264,8 @@ public class WebAppClassLoader extends URLClassLoader
{
Resource fn=lib.addPath(files[f]);
String fnlc=fn.getName().toLowerCase();
if (!fn.isDirectory() && isFileSupported(fnlc))
// don't check if this is a directory, see Bug 353165
if (isFileSupported(fnlc))
{
String jar=fn.toString();
jar=StringUtil.replace(jar, ",", "%2C");

View File

@ -111,6 +111,7 @@ public class WebAppContext extends ServletContextHandler implements WebAppClassL
"-org.eclipse.jetty.plus.jaas.", // don't hide jaas classes
"-org.eclipse.jetty.websocket.", // don't hide websocket extension
"-org.eclipse.jetty.servlet.DefaultServlet", // don't hide default servlet
"-org.eclipse.jetty.servlet.listener.", //don't hide useful listeners
"org.eclipse.jetty." // hide other jetty classes
} ;

View File

@ -82,7 +82,7 @@ public class DeflateFrameExtension extends AbstractExtension
catch(DataFormatException e)
{
Log.warn(e);
getConnection().close(WebSocketConnectionD7_9.CLOSE_PROTOCOL,e.toString());
getConnection().close(WebSocketConnectionD10.CLOSE_PROTOCOL,e.toString());
}
}

View File

@ -5,12 +5,19 @@ import java.io.BufferedWriter;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.URI;
import java.security.SecureRandom;
import java.util.Arrays;
import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import org.eclipse.jetty.io.Buffer;
import org.eclipse.jetty.io.bio.SocketEndPoint;
@ -18,6 +25,8 @@ import org.eclipse.jetty.util.B64Code;
import org.eclipse.jetty.util.StringUtil;
import org.eclipse.jetty.util.TypeUtil;
import org.eclipse.jetty.util.log.Log;
import org.eclipse.jetty.websocket.WebSocket.Connection;
import org.eclipse.jetty.websocket.WebSocket.FrameConnection;
/**
* @version $Revision$ $Date$
@ -25,250 +34,149 @@ import org.eclipse.jetty.util.log.Log;
* This is not a general purpose websocket client.
* It's only for testing the websocket server and is hardwired to a specific draft version of the protocol.
*/
public class TestClient
public class TestClient implements WebSocket.OnFrame
{
private final static Random __random = new SecureRandom();
private static WebSocketClient __client = new WebSocketClient();
private static boolean _verbose=false;
private static final Random __random = new Random();
private final String _host;
private final int _port;
private final String _protocol;
private int _size=64;
private final Socket _socket;
private final BufferedWriter _output;
private final BufferedReader _input;
private final SocketEndPoint _endp;
private final WebSocketGeneratorD7_9 _generator;
private final WebSocketParserD7_9 _parser;
private int _framesSent;
private int _messagesSent;
private int _framesReceived;
private int _messagesReceived;
private long _totalTime;
private long _minDuration=Long.MAX_VALUE;
private long _maxDuration=Long.MIN_VALUE;
private long _start;
private final int _timeout;
private static boolean __quiet;
private static int __framesSent;
private static int __messagesSent;
private static AtomicInteger __framesReceived=new AtomicInteger();
private static AtomicInteger __messagesReceived=new AtomicInteger();
private static AtomicLong __totalTime=new AtomicLong();
private static AtomicLong __minDuration=new AtomicLong(Long.MAX_VALUE);
private static AtomicLong __maxDuration=new AtomicLong(Long.MIN_VALUE);
private static long __start;
private BlockingQueue<Long> _starts = new LinkedBlockingQueue<Long>();
int _messageBytes;
int _frames;
byte _opcode=-1;
private final WebSocketParser.FrameHandler _handler = new WebSocketParser.FrameHandler()
private volatile WebSocket.FrameConnection _connection;
private final CountDownLatch _handshook = new CountDownLatch(1);
public void onOpen(Connection connection)
{
public synchronized void onFrame(byte flags, byte opcode, Buffer buffer)
}
public void onError(String message, Throwable ex)
{
System.err.println("onError: "+message);
if (ex!=null)
ex.printStackTrace();
_handshook.countDown();
}
public void onClose(int closeCode, String message)
{
_handshook.countDown();
}
public boolean onFrame(byte flags, byte opcode, byte[] data, int offset, int length)
{
try
{
try
{
_framesReceived++;
_frames++;
if (opcode == WebSocketConnectionD7_9.OP_CLOSE)
{
byte[] data=buffer.asArray();
// System.err.println("CLOSED: "+((0xff&data[0])*0x100+(0xff&data[1]))+" "+new String(data,2,data.length-2,StringUtil.__UTF8));
_generator.addFrame((byte)0x8,WebSocketConnectionD7_9.OP_CLOSE,data,0,data.length);
_generator.flush();
_socket.shutdownOutput();
_socket.close();
return;
}
else if (opcode == WebSocketConnectionD7_9.OP_PING)
{
_generator.addFrame((byte)0x8,WebSocketConnectionD7_9.OP_PONG,buffer.array(),buffer.getIndex(),buffer.length());
_generator.flush();
}
_messageBytes+=buffer.length();
if (_opcode==-1)
_opcode=opcode;
if (_connection.isClose(opcode))
return false;
__framesReceived.incrementAndGet();
_frames++;
_messageBytes+=length;
if (_opcode==-1)
_opcode=opcode;
if (_connection.isControl(opcode) || _connection.isMessageComplete(flags))
{
int recv =__messagesReceived.incrementAndGet();
Long start=_starts.poll();
if (WebSocketConnectionD7_9.isLastFrame(flags))
if (start!=null)
{
_messagesReceived++;
Long start=_starts.take();
long duration = System.nanoTime()-start.longValue();
if (duration>_maxDuration)
_maxDuration=duration;
if (duration<_minDuration)
_minDuration=duration;
_totalTime+=duration;
System.out.printf("%d bytes from %s: frames=%d req=%d time=%.1fms opcode=0x%s\n",_messageBytes,_host,_frames,_messagesReceived,((double)duration/1000000.0),TypeUtil.toHexString(_opcode));
_frames=0;
_messageBytes=0;
_opcode=-1;
long max=__maxDuration.get();
while(duration>max && !__maxDuration.compareAndSet(max,duration))
max=__maxDuration.get();
long min=__minDuration.get();
while(duration<min && !__minDuration.compareAndSet(min,duration))
min=__minDuration.get();
__totalTime.addAndGet(duration);
if (!__quiet)
System.out.printf("%d bytes from %s: frames=%d req=%d time=%.1fms opcode=0x%s\n",_messageBytes,_host,_frames,recv,((double)duration/1000000.0),TypeUtil.toHexString(_opcode));
}
_frames=0;
_messageBytes=0;
_opcode=-1;
}
catch(Exception e)
{
e.printStackTrace();
}
}
public void close(int code,String message)
catch(Exception e)
{
e.printStackTrace();
}
};
return false;
}
public void onHandshake(FrameConnection connection)
{
_connection=connection;
_handshook.countDown();
}
public TestClient(String host, int port,String protocol, int timeoutMS) throws IOException
public TestClient(String host, int port,String protocol, int timeoutMS) throws Exception
{
_host=host;
_port=port;
_protocol=protocol;
_socket = new Socket(host, port);
_socket.setSoTimeout(timeoutMS);
_output = new BufferedWriter(new OutputStreamWriter(_socket.getOutputStream(), "ISO-8859-1"));
_input = new BufferedReader(new InputStreamReader(_socket.getInputStream(), "ISO-8859-1"));
_endp=new SocketEndPoint(_socket);
_generator = new WebSocketGeneratorD7_9(new WebSocketBuffers(32*1024),_endp,new WebSocketGeneratorD7_9.FixedMaskGen(new byte[4]));
_parser = new WebSocketParserD7_9(new WebSocketBuffers(32*1024),_endp,_handler,false);
_timeout=timeoutMS;
}
public int getSize()
private void open() throws Exception
{
return _size;
__client.open(new URI("ws://"+_host+":"+_port+"/"),this,_protocol,_timeout);
_handshook.await(10,TimeUnit.SECONDS);
}
public void setSize(int size)
public void ping(byte opcode,byte[] data,int fragment) throws Exception
{
_size = size;
}
_starts.add(System.nanoTime());
private void open() throws IOException
{
System.out.println("Jetty WebSocket PING "+_host+":"+_port+
" ("+_socket.getRemoteSocketAddress()+") " +_size+" bytes of data.");
byte[] key = new byte[16];
__random.nextBytes(key);
_output.write("GET /chat HTTP/1.1\r\n"+
"Host: "+_host+":"+_port+"\r\n"+
"Upgrade: websocket\r\n"+
"Connection: Upgrade\r\n"+
"Sec-WebSocket-Key: "+new String(B64Code.encode(key))+"\r\n"+
"Sec-WebSocket-Origin: http://example.com\r\n"+
"Sec-WebSocket-Protocol: "+_protocol+"\r\n" +
"Sec-WebSocket-Version: 7\r\n"+
"\r\n");
_output.flush();
int off=0;
int len=data.length;
if (fragment>0&& len>fragment)
len=fragment;
__messagesSent++;
while(off<data.length)
{
__framesSent++;
byte flags= (byte)(off+len==data.length?0x8:0);
byte op=(byte)(off==0?opcode:WebSocketConnectionD10.OP_CONTINUATION);
String responseLine = _input.readLine();
if(!responseLine.startsWith("HTTP/1.1 101 Switching Protocols"))
throw new IOException(responseLine);
// Read until we find Response key
String line;
boolean accepted=false;
String protocol="";
while ((line = _input.readLine()) != null)
{
if (line.length() == 0)
break;
if (line.startsWith("Sec-WebSocket-Accept:"))
{
String accept=line.substring(21).trim();
accepted=accept.equals(WebSocketConnectionD7_9.hashKey(new String(B64Code.encode(key))));
}
else if (line.startsWith("Sec-WebSocket-Protocol:"))
{
protocol=line.substring(24).trim();
}
}
if (!accepted)
throw new IOException("Bad Sec-WebSocket-Accept");
System.out.println("handshake OK for protocol '"+protocol+"'");
new Thread()
{
public void run()
{
while (_endp.isOpen())
{
_parser.parseNext();
}
}
}.start();
}
if (_verbose)
System.err.printf("%s#addFrame %s|%s %s\n",this.getClass().getSimpleName(),TypeUtil.toHexString(flags),TypeUtil.toHexString(op),TypeUtil.toHexString(data,off,len));
public void ping(int count,byte opcode,int fragment)
{
try
{
_start=System.currentTimeMillis();
for (int i=0;i<count && !_socket.isClosed();i++)
{
if (_socket.isClosed())
break;
byte data[]=null;
_connection.sendFrame(flags,op,data,off,len);
if (opcode==WebSocketConnectionD7_9.OP_TEXT)
{
StringBuilder b = new StringBuilder();
while (b.length()<_size)
b.append('A'+__random.nextInt(26));
data=b.toString().getBytes(StringUtil.__UTF8);
}
else
{
data= new byte[_size];
__random.nextBytes(data);
}
_starts.add(System.nanoTime());
int off=0;
int len=data.length;
if (fragment>0&& len>fragment)
len=fragment;
_messagesSent++;
while(off<data.length)
{
_framesSent++;
byte flags= (byte)(off+len==data.length?0x8:0);
byte op=(byte)(off==0?opcode:WebSocketConnectionD7_9.OP_CONTINUATION);
if (_verbose)
System.err.printf("%s#addFrame %s|%s %s\n",this.getClass().getSimpleName(),TypeUtil.toHexString(flags),TypeUtil.toHexString(op),TypeUtil.toHexString(data,off,len));
_generator.addFrame(flags,op,data,off,len);
off+=len;
if(data.length-off>len)
len=data.length-off;
if (fragment>0&& len>fragment)
len=fragment;
}
_generator.flush();
Thread.sleep(1000);
}
}
catch (Exception x)
{
throw new RuntimeException(x);
off+=len;
if(data.length-off>len)
len=data.length-off;
if (fragment>0&& len>fragment)
len=fragment;
}
}
public void dump() throws Exception
public void disconnect() throws Exception
{
for (int i=0;i<250;i++)
{
if (_messagesSent==_messagesReceived)
break;
_generator.flush();
Thread.sleep(100);
}
_socket.close();
long duration=System.currentTimeMillis()-_start;
System.out.println("--- "+_host+" websocket ping statistics using 1 connection ---");
System.out.println(_framesSent+" frames transmitted, "+_framesReceived+" received, "+
_messagesSent+" messages transmitted, "+_messagesReceived+" received, "+
"time "+duration+"ms");
System.out.printf("rtt min/ave/max = %.3f/%.3f/%.3f ms\n",_minDuration/1000000.0,_messagesReceived==0?0.0:(_totalTime/_messagesReceived/1000000.0),_maxDuration/1000000.0);
if (_connection!=null)
_connection.disconnect();
}
@ -284,66 +192,114 @@ public class TestClient
System.err.println(" -s|--size n (default 64)");
System.err.println(" -f|--fragment n (default 4000) ");
System.err.println(" -P|--protocol echo|echo-assemble|echo-fragment|echo-broadcast");
System.err.println(" -C|--clients n (default 1) ");
System.err.println(" -d|--delay n (default 1000ms) ");
System.exit(1);
}
public static void main(String[] args)
public static void main(String[] args) throws Exception
{
__client.start();
String host="localhost";
int port=8080;
String protocol=null;
int count=10;
int size=64;
int fragment=4000;
boolean binary=false;
int clients=1;
int delay=1000;
for (int i=0;i<args.length;i++)
{
String a=args[i];
if ("-p".equals(a)||"--port".equals(a))
port=Integer.parseInt(args[++i]);
else if ("-h".equals(a)||"--host".equals(a))
port=Integer.parseInt(args[++i]);
else if ("-c".equals(a)||"--count".equals(a))
count=Integer.parseInt(args[++i]);
else if ("-s".equals(a)||"--size".equals(a))
size=Integer.parseInt(args[++i]);
else if ("-f".equals(a)||"--fragment".equals(a))
fragment=Integer.parseInt(args[++i]);
else if ("-P".equals(a)||"--protocol".equals(a))
protocol=args[++i];
else if ("-v".equals(a)||"--verbose".equals(a))
_verbose=true;
else if ("-b".equals(a)||"--binary".equals(a))
binary=true;
else if ("-C".equals(a)||"--clients".equals(a))
clients=Integer.parseInt(args[++i]);
else if ("-d".equals(a)||"--delay".equals(a))
delay=Integer.parseInt(args[++i]);
else if ("-q".equals(a)||"--quiet".equals(a))
__quiet=true;
else if (a.startsWith("-"))
usage(args);
}
TestClient[] client = new TestClient[clients];
try
{
String host="localhost";
int port=8080;
String protocol=null;
int count=10;
int size=64;
int fragment=4000;
boolean binary=false;
for (int i=0;i<args.length;i++)
__start=System.currentTimeMillis();
for (int i=0;i<clients;i++)
{
String a=args[i];
if ("-p".equals(a)||"--port".equals(a))
port=Integer.parseInt(args[++i]);
else if ("-h".equals(a)||"--host".equals(a))
port=Integer.parseInt(args[++i]);
else if ("-c".equals(a)||"--count".equals(a))
count=Integer.parseInt(args[++i]);
else if ("-s".equals(a)||"--size".equals(a))
size=Integer.parseInt(args[++i]);
else if ("-f".equals(a)||"--fragment".equals(a))
fragment=Integer.parseInt(args[++i]);
else if ("-P".equals(a)||"--protocol".equals(a))
protocol=args[++i];
else if ("-v".equals(a)||"--verbose".equals(a))
_verbose=true;
else if ("-b".equals(a)||"--binary".equals(a))
binary=true;
else if (a.startsWith("-"))
usage(args);
client[i]=new TestClient(host,port,protocol==null?null:protocol,10000);
client[i].open();
}
System.out.println("Jetty WebSocket PING "+host+":"+port+
" ("+ new InetSocketAddress(host,port)+") "+clients+" clients");
TestClient client = new TestClient(host,port,protocol==null?null:protocol,10000);
client.setSize(size);
try
for (int p=0;p<count;p++)
{
client.open();
if (protocol!=null && protocol.startsWith("echo"))
client.ping(count,binary?WebSocketConnectionD7_9.OP_BINARY:WebSocketConnectionD7_9.OP_TEXT,fragment);
long next = System.currentTimeMillis()+delay;
byte opcode=binary?WebSocketConnectionD10.OP_BINARY:WebSocketConnectionD10.OP_TEXT;
byte data[]=null;
if (opcode==WebSocketConnectionD10.OP_TEXT)
{
StringBuilder b = new StringBuilder();
while (b.length()<size)
b.append('A'+__random.nextInt(26));
data=b.toString().getBytes(StringUtil.__UTF8);
}
else
client.ping(count,WebSocketConnectionD7_9.OP_PING,-1);
{
data= new byte[size];
__random.nextBytes(data);
}
for (int i=0;i<clients;i++)
client[i].ping(opcode,data,opcode==WebSocketConnectionD10.OP_PING?-1:fragment);
while(System.currentTimeMillis()<next)
Thread.sleep(10);
}
finally
{
client.dump();
}
}
catch (Exception e)
finally
{
Log.warn(e);
for (int i=0;i<clients;i++)
if (client[i]!=null)
client[i].disconnect();
long duration=System.currentTimeMillis()-__start;
System.out.println("--- "+host+" websocket ping statistics using "+clients+" connection"+(clients>1?"s":"")+" ---");
System.out.println(__framesSent+" frames transmitted, "+__framesReceived+" received, "+
__messagesSent+" messages transmitted, "+__messagesReceived+" received, "+
"time "+duration+"ms "+ (1000L*__messagesReceived.get()/duration)+" req/s");
System.out.printf("rtt min/ave/max = %.3f/%.3f/%.3f ms\n",__minDuration.get()/1000000.0,__messagesReceived.get()==0?0.0:(__totalTime.get()/__messagesReceived.get()/1000000.0),__maxDuration.get()/1000000.0);
__client.stop();
}
}

View File

@ -107,6 +107,12 @@ public class TestServer extends Server
System.err.printf("%s#onOpen %s\n",this.getClass().getSimpleName(),connection);
}
public void onError(String message, Throwable ex)
{
if (_verbose)
System.err.printf("%s#onOpen %s\n",this.getClass().getSimpleName(),message);
}
public void onHandshake(FrameConnection connection)
{
if (_verbose)

View File

@ -28,6 +28,13 @@ public interface WebSocket
* @param connection The Connection object to use to send messages.
*/
void onOpen(Connection connection);
/**
* Called when a new websocket connection cannot be created
* @param message The error message
* @param ex The exception or null
*/
void onError(String message, Throwable ex);
/**
* Called when an established websocket connection closes

View File

@ -34,7 +34,7 @@ public class WebSocketBuffers
{
final private int _bufferSize;
final private Buffers _buffers;
final private int _maxBuffers=1024;
final private int _maxBuffers=-1;
public WebSocketBuffers(final int bufferSize)
{

View File

@ -0,0 +1,497 @@
package org.eclipse.jetty.websocket;
import java.io.EOFException;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.URI;
import java.nio.channels.ByteChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.nio.channels.UnsupportedAddressTypeException;
import java.util.List;
import java.util.Map;
import java.util.Random;
import org.eclipse.jetty.http.HttpFields;
import org.eclipse.jetty.http.HttpParser;
import org.eclipse.jetty.io.AbstractConnection;
import org.eclipse.jetty.io.Buffer;
import org.eclipse.jetty.io.Buffers;
import org.eclipse.jetty.io.ByteArrayBuffer;
import org.eclipse.jetty.io.ConnectedEndPoint;
import org.eclipse.jetty.io.Connection;
import org.eclipse.jetty.io.SimpleBuffers;
import org.eclipse.jetty.io.nio.SelectChannelEndPoint;
import org.eclipse.jetty.io.nio.SelectorManager;
import org.eclipse.jetty.util.B64Code;
import org.eclipse.jetty.util.QuotedStringTokenizer;
import org.eclipse.jetty.util.component.AggregateLifeCycle;
import org.eclipse.jetty.util.log.Logger;
import org.eclipse.jetty.util.thread.QueuedThreadPool;
import org.eclipse.jetty.util.thread.ThreadPool;
import org.eclipse.jetty.util.thread.Timeout;
public class WebSocketClient extends AggregateLifeCycle
{
private final static Logger __log = org.eclipse.jetty.util.log.Log.getLogger(WebSocketClient.class.getCanonicalName());
private final static Random __random = new Random();
private final static ByteArrayBuffer __ACCEPT = new ByteArrayBuffer.CaseInsensitive("Sec-WebSocket-Accept");
private final ThreadPool _threadPool;
private final Selector _selector=new Selector();
private final Timeout _connectQ=new Timeout();
private int _connectTimeout=30000;
private int _bufferSize=64*1024;
private boolean _blockingConnect=false;
private WebSocketBuffers _buffers;
public WebSocketClient(ThreadPool threadpool)
{
_threadPool=threadpool;
addBean(_selector);
addBean(_threadPool);
}
public WebSocketClient()
{
this(new QueuedThreadPool());
}
public SelectorManager getSelectorManager()
{
return _selector;
}
public ThreadPool getThreadPool()
{
return _threadPool;
}
public int getConnectTimeout()
{
return _connectTimeout;
}
public void setConnectTimeout(int connectTimeout)
{
if (isRunning())
throw new IllegalStateException(getState());
_connectTimeout = connectTimeout;
}
public int getMaxIdleTime()
{
return (int)_selector.getMaxIdleTime();
}
public void setMaxIdleTime(int maxIdleTime)
{
_selector.setMaxIdleTime(maxIdleTime);
}
public int getBufferSize()
{
return _bufferSize;
}
public void setBufferSize(int bufferSize)
{
_bufferSize = bufferSize;
}
public boolean isBlockingConnect()
{
return _blockingConnect;
}
public void setBlockingConnect(boolean blockingConnect)
{
_blockingConnect = blockingConnect;
}
@Override
protected void doStart() throws Exception
{
_buffers = new WebSocketBuffers(_bufferSize);
super.doStart();
for (int i=0;i<_selector.getSelectSets();i++)
{
final int id=i;
_threadPool.dispatch(new Runnable(){
public void run()
{
while(isRunning())
{
try
{
_selector.doSelect(id);
}
catch (IOException e)
{
__log.warn(e);
}
}
}
});
}
_connectQ.setDuration(_connectTimeout);
_threadPool.dispatch(new Runnable(){
public void run()
{
while(isRunning())
{
try
{
Thread.sleep(200); // TODO configure?
_connectQ.tick(System.currentTimeMillis());
}
catch(Exception e)
{
__log.warn(e);
}
}
}
});
}
public void open(URI uri, WebSocket websocket) throws IOException
{
open(uri,websocket,null,(int)_selector.getMaxIdleTime(),null,null);
}
public void open(URI uri, WebSocket websocket, String protocol,int maxIdleTime) throws IOException
{
open(uri,websocket,protocol,(int)_selector.getMaxIdleTime(),null,null);
}
public void open(URI uri, WebSocket websocket, String protocol,int maxIdleTime,Map<String,String> cookies) throws IOException
{
open(uri,websocket,protocol,(int)_selector.getMaxIdleTime(),cookies,null);
}
public void open(URI uri, WebSocket websocket, String protocol,int maxIdleTime,Map<String,String> cookies,List<String> extensions) throws IOException
{
if (!isStarted())
throw new IllegalStateException("!started");
String scheme=uri.getScheme();
if (!("ws".equalsIgnoreCase(scheme) || "wss".equalsIgnoreCase(scheme)))
throw new IllegalArgumentException("Bad WebSocket scheme '"+scheme+"'");
if ("wss".equalsIgnoreCase(scheme))
throw new IOException("wss not supported");
SocketChannel channel = SocketChannel.open();
channel.socket().setTcpNoDelay(true);
channel.socket().setSoTimeout(getMaxIdleTime());
InetSocketAddress address=new InetSocketAddress(uri.getHost(),uri.getPort());
WebSocketHolder holder=new WebSocketHolder(websocket,uri,protocol,maxIdleTime,cookies,extensions,channel);
_connectQ.schedule(holder);
boolean thrown=true;
try
{
if (isBlockingConnect())
{
channel.socket().connect(address,0);
channel.configureBlocking(false);
}
else
{
channel.configureBlocking(false);
channel.connect(address);
}
_selector.register( channel, holder);
thrown=false;
}
finally
{
if (thrown)
holder.cancel();
}
}
class Selector extends SelectorManager
{
@Override
public boolean dispatch(Runnable task)
{
return _threadPool.dispatch(task);
}
@Override
protected SelectChannelEndPoint newEndPoint(SocketChannel channel, SelectSet selectSet, final SelectionKey sKey) throws IOException
{
return new SelectChannelEndPoint(channel,selectSet,sKey);
}
@Override
protected Connection newConnection(SocketChannel channel, SelectChannelEndPoint endpoint)
{
WebSocketHolder holder = (WebSocketHolder) endpoint.getSelectionKey().attachment();
return new HandshakeConnection(endpoint,holder);
}
@Override
protected void endPointOpened(SelectChannelEndPoint endpoint)
{
}
@Override
protected void endPointUpgraded(ConnectedEndPoint endpoint, Connection oldConnection)
{
}
@Override
protected void endPointClosed(SelectChannelEndPoint endpoint)
{
endpoint.getConnection().closed();
}
@Override
protected void connectionFailed(SocketChannel channel, Throwable ex, Object attachment)
{
if (!(attachment instanceof WebSocketHolder))
super.connectionFailed(channel,ex,attachment);
else
{
__log.debug(ex);
WebSocketHolder holder = (WebSocketHolder)attachment;
holder.cancel();
holder.getWebSocket().onError(ex.toString(),ex);
}
}
}
class HandshakeConnection extends AbstractConnection
{
private final SelectChannelEndPoint _endp;
private final WebSocketHolder _holder;
private final String _key;
private final HttpParser _parser;
private String _accept;
private String _error;
public HandshakeConnection(SelectChannelEndPoint endpoint, WebSocketHolder holder)
{
super(endpoint,System.currentTimeMillis());
_endp=endpoint;
_holder=holder;
byte[] bytes=new byte[16];
__random.nextBytes(bytes);
_key=new String(B64Code.encode(bytes));
Buffers buffers = new SimpleBuffers(_buffers.getBuffer(),null);
_parser=new HttpParser(buffers,_endp,
new HttpParser.EventHandler()
{
@Override
public void startResponse(Buffer version, int status, Buffer reason) throws IOException
{
if (status!=101)
{
_error="Bad response status "+status+" "+reason;
_endp.close();
}
}
@Override
public void parsedHeader(Buffer name, Buffer value) throws IOException
{
if (__ACCEPT.equals(name))
_accept=value.toString();
}
@Override
public void startRequest(Buffer method, Buffer url, Buffer version) throws IOException
{
if (_error==null)
_error="Bad response: "+method+" "+url+" "+version;
_endp.close();
}
@Override
public void content(Buffer ref) throws IOException
{
if (_error==null)
_error="Bad response. "+ref.length()+"B of content?";
_endp.close();
}
});
String request=
"GET "+_holder.getURI().getPath()+" HTTP/1.1\r\n"+
"Host: "+holder.getURI().getHost()+":"+_holder.getURI().getPort()+"\r\n"+
"Upgrade: websocket\r\n"+
"Connection: Upgrade\r\n"+
"Sec-WebSocket-Key: "+_key+"\r\n"+
"Sec-WebSocket-Origin: http://example.com\r\n"+
"Sec-WebSocket-Version: 8\r\n";
if (holder.getProtocol()!=null)
request+="Sec-WebSocket-Protocol: "+holder.getProtocol()+"\r\n";
if (holder.getCookies()!=null && holder.getCookies().size()>0)
{
for (String cookie : holder.getCookies().keySet())
request+="Cookie: "+QuotedStringTokenizer.quoteIfNeeded(cookie,HttpFields.__COOKIE_DELIM)+
"="+
QuotedStringTokenizer.quoteIfNeeded(holder.getCookies().get(cookie),HttpFields.__COOKIE_DELIM)+
"\r\n";
}
request+="\r\n";
// TODO extensions
try
{
ByteArrayBuffer handshake = new ByteArrayBuffer(request);
int len=handshake.length();
if (len!=_endp.flush(handshake))
throw new IOException("incomplete");
}
catch(IOException e)
{
__log.debug(e);
_holder.getWebSocket().onError("Handshake failed",e);
}
}
public Connection handle() throws IOException
{
while (_endp.isOpen() && !_parser.isComplete())
{
switch (_parser.parseAvailable())
{
case -1:
_holder.cancel();
_holder.getWebSocket().onError("EOF",new EOFException());
return this;
case 0:
return this;
default:
break;
}
}
if (_error==null && _accept==null)
_error="No Sec-WebSocket-Accept";
else if (_error==null && !WebSocketConnectionD10.hashKey(_key).equals(_accept))
_error="Bad Sec-WebSocket-Accept";
else
{
Buffer header=_parser.getHeaderBuffer();
WebSocketConnectionD10 connection = new WebSocketConnectionD10(_holder.getWebSocket(),_endp,_buffers,System.currentTimeMillis(),_holder.getMaxIdleTime(),_holder.getProtocol(),null,10, new WebSocketGeneratorD10.RandomMaskGen());
if (header.hasContent())
connection.fillBuffersFrom(header);
_buffers.returnBuffer(header);
if (_holder.getWebSocket() instanceof WebSocket.OnFrame)
((WebSocket.OnFrame)_holder.getWebSocket()).onHandshake((WebSocket.FrameConnection)connection.getConnection());
_holder.cancel();
_holder.getWebSocket().onOpen(connection.getConnection());
return connection;
}
_endp.close();
return this;
}
public boolean isIdle()
{
return false;
}
public boolean isSuspended()
{
return false;
}
public void closed()
{
_holder.cancel();
_holder.getWebSocket().onError(_error==null?"EOF":_error,null);
}
}
class WebSocketHolder extends Timeout.Task
{
final WebSocket _websocket;;
final URI _uri;
final String _protocol;
final int _maxIdleTime;
final Map<String,String> _cookies;
final List<String> _extensions;
final ByteChannel _channel;
public WebSocketHolder(WebSocket websocket, URI uri, String protocol, int maxIdleTime, Map<String,String> cookies,List<String> extensions, ByteChannel channel)
{
_websocket=websocket;
_uri=uri;
_protocol=protocol;
_maxIdleTime=maxIdleTime;
_cookies=cookies;
_extensions=extensions;
_channel=channel;
}
public Map<String,String> getCookies()
{
return _cookies;
}
public String getProtocol()
{
return _protocol;
}
public WebSocket getWebSocket()
{
return _websocket;
}
public URI getURI()
{
return _uri;
}
public int getMaxIdleTime()
{
return _maxIdleTime;
}
@Override
public void expired()
{
try
{
__log.debug("expired "+this);
getWebSocket().onError("expired",null);
_channel.close();
}
catch(IOException e)
{
__log.ignore(e);
}
}
public String toString()
{
return "[" + _uri + ","+_websocket+"]@"+hashCode();
}
}
}

View File

@ -37,8 +37,9 @@ import org.eclipse.jetty.websocket.WebSocket.OnFrame;
import org.eclipse.jetty.websocket.WebSocket.OnTextMessage;
import org.eclipse.jetty.websocket.WebSocket.OnBinaryMessage;
import org.eclipse.jetty.websocket.WebSocket.OnControl;
import org.eclipse.jetty.websocket.WebSocketGeneratorD10.MaskGen;
public class WebSocketConnectionD7_9 extends AbstractConnection implements WebSocketConnection
public class WebSocketConnectionD10 extends AbstractConnection implements WebSocketConnection
{
final static byte OP_CONTINUATION = 0x00;
final static byte OP_TEXT = 0x01;
@ -55,6 +56,9 @@ public class WebSocketConnectionD7_9 extends AbstractConnection implements WebSo
final static int CLOSE_PROTOCOL=1002;
final static int CLOSE_BADDATA=1003;
final static int CLOSE_LARGE=1004;
final static int CLOSE_NOCODE=1005;
final static int CLOSE_NOCLOSE=1006;
final static int CLOSE_NOTUTF8=1007;
static boolean isLastFrame(byte flags)
{
@ -69,9 +73,9 @@ public class WebSocketConnectionD7_9 extends AbstractConnection implements WebSo
private final static byte[] MAGIC;
private final IdleCheck _idle;
private final List<Extension> _extensions;
private final WebSocketParserD7_9 _parser;
private final WebSocketParserD10 _parser;
private final WebSocketParser.FrameHandler _inbound;
private final WebSocketGeneratorD7_9 _generator;
private final WebSocketGeneratorD10 _generator;
private final WebSocketGenerator _outbound;
private final WebSocket _webSocket;
private final OnFrame _onFrame;
@ -80,6 +84,7 @@ public class WebSocketConnectionD7_9 extends AbstractConnection implements WebSo
private final OnControl _onControl;
private final String _protocol;
private final int _draft;
private int _close;
private boolean _closedIn;
private boolean _closedOut;
private int _maxTextMessageSize;
@ -103,11 +108,18 @@ public class WebSocketConnectionD7_9 extends AbstractConnection implements WebSo
/* ------------------------------------------------------------ */
/* ------------------------------------------------------------ */
/* ------------------------------------------------------------ */
private final WebSocket.FrameConnection _connection = new FrameConnectionD07();
private final WebSocket.FrameConnection _connection = new FrameConnectionD10();
/* ------------------------------------------------------------ */
public WebSocketConnectionD7_9(WebSocket websocket, EndPoint endpoint, WebSocketBuffers buffers, long timestamp, int maxIdleTime, String protocol, List<Extension> extensions,int draft)
public WebSocketConnectionD10(WebSocket websocket, EndPoint endpoint, WebSocketBuffers buffers, long timestamp, int maxIdleTime, String protocol, List<Extension> extensions,int draft)
throws IOException
{
this(websocket,endpoint,buffers,timestamp,maxIdleTime,protocol,extensions,draft,null);
}
/* ------------------------------------------------------------ */
public WebSocketConnectionD10(WebSocket websocket, EndPoint endpoint, WebSocketBuffers buffers, long timestamp, int maxIdleTime, String protocol, List<Extension> extensions,int draft, MaskGen maskgen)
throws IOException
{
super(endpoint,timestamp);
@ -124,7 +136,7 @@ public class WebSocketConnectionD7_9 extends AbstractConnection implements WebSo
_onTextMessage=_webSocket instanceof OnTextMessage ? (OnTextMessage)_webSocket : null;
_onBinaryMessage=_webSocket instanceof OnBinaryMessage ? (OnBinaryMessage)_webSocket : null;
_onControl=_webSocket instanceof OnControl ? (OnControl)_webSocket : null;
_generator = new WebSocketGeneratorD7_9(buffers, _endp,null);
_generator = new WebSocketGeneratorD10(buffers, _endp,maskgen);
_extensions=extensions;
if (_extensions!=null)
@ -140,10 +152,10 @@ public class WebSocketConnectionD7_9 extends AbstractConnection implements WebSo
}
}
_outbound=_extensions.size()==0?_generator:extensions.get(extensions.size()-1);
_inbound=_extensions.size()==0?_frameHandler:extensions.get(0);
_outbound=(_extensions==null||_extensions.size()==0)?_generator:extensions.get(extensions.size()-1);
_inbound=(_extensions==null||_extensions.size()==0)?_frameHandler:extensions.get(0);
_parser = new WebSocketParserD7_9(buffers, endpoint,_inbound,true);
_parser = new WebSocketParserD10(buffers, endpoint,_inbound,maskgen==null);
_protocol=protocol;
@ -232,7 +244,7 @@ public class WebSocketConnectionD7_9 extends AbstractConnection implements WebSo
if (_closedIn && _closedOut && _outbound.isBufferEmpty())
_endp.close();
else if (_endp.isInputShutdown() && !_closedIn)
closeIn(CLOSE_PROTOCOL,null);
closeIn(CLOSE_NOCLOSE,null);
else
checkWriteable();
}
@ -251,7 +263,7 @@ public class WebSocketConnectionD7_9 extends AbstractConnection implements WebSo
@Override
public void idleExpired()
{
closeOut(WebSocketConnectionD7_9.CLOSE_NORMAL,"Idle");
closeOut(WebSocketConnectionD10.CLOSE_NORMAL,"Idle");
}
/* ------------------------------------------------------------ */
@ -263,57 +275,98 @@ public class WebSocketConnectionD7_9 extends AbstractConnection implements WebSo
/* ------------------------------------------------------------ */
public void closed()
{
_webSocket.onClose(WebSocketConnectionD7_9.CLOSE_NORMAL,"");
final boolean closed;
synchronized (this)
{
closed=_close==0;
if (closed)
_close=WebSocketConnectionD10.CLOSE_NOCLOSE;
}
if (closed)
_webSocket.onClose(WebSocketConnectionD10.CLOSE_NOCLOSE,"closed");
}
/* ------------------------------------------------------------ */
public synchronized void closeIn(int code,String message)
public void closeIn(int code,String message)
{
Log.debug("ClosedIn {} {}",this,message);
final boolean closedOut;
final boolean closed;
synchronized (this)
{
closedOut=_closedOut;
_closedIn=true;
closed=_close==0;
if (closed)
_close=code;
}
try
{
if (_closedOut)
_endp.close();
else
closeOut(code,message);
}
catch(IOException e)
{
Log.ignore(e);
if (closed)
_webSocket.onClose(code,message);
}
finally
{
_closedIn=true;
try
{
if (closedOut)
_endp.close();
else
closeOut(code,message);
}
catch(IOException e)
{
Log.ignore(e);
}
}
}
/* ------------------------------------------------------------ */
public synchronized void closeOut(int code,String message)
public void closeOut(int code,String message)
{
Log.debug("ClosedOut {} {}",this,message);
final boolean close;
final boolean closed;
synchronized (this)
{
close=_closedIn || _closedOut;
_closedOut=true;
closed=_close==0;
if (closed)
_close=code;
}
try
{
if (_closedIn || _closedOut)
_endp.close();
else
{
if (code<=0)
code=WebSocketConnectionD7_9.CLOSE_NORMAL;
byte[] bytes = ("xx"+(message==null?"":message)).getBytes(StringUtil.__ISO_8859_1);
bytes[0]=(byte)(code/0x100);
bytes[1]=(byte)(code%0x100);
_outbound.addFrame((byte)0x8,WebSocketConnectionD7_9.OP_CLOSE,bytes,0,bytes.length);
}
_outbound.flush();
}
catch(IOException e)
{
Log.ignore(e);
if (closed)
_webSocket.onClose(code,message);
}
finally
{
_closedOut=true;
try
{
if (close)
_endp.close();
else
{
if (code<=0)
code=WebSocketConnectionD10.CLOSE_NORMAL;
byte[] bytes = ("xx"+(message==null?"":message)).getBytes(StringUtil.__ISO_8859_1);
bytes[0]=(byte)(code/0x100);
bytes[1]=(byte)(code%0x100);
_outbound.addFrame((byte)0x8,WebSocketConnectionD10.OP_CLOSE,bytes,0,bytes.length);
}
_outbound.flush();
}
catch(IOException e)
{
Log.ignore(e);
}
}
}
@ -335,11 +388,11 @@ public class WebSocketConnectionD7_9 extends AbstractConnection implements WebSo
/* ------------------------------------------------------------ */
/* ------------------------------------------------------------ */
/* ------------------------------------------------------------ */
private class FrameConnectionD07 implements WebSocket.FrameConnection
private class FrameConnectionD10 implements WebSocket.FrameConnection
{
volatile boolean _disconnecting;
int _maxTextMessage=WebSocketConnectionD7_9.this._maxTextMessageSize;
int _maxBinaryMessage=WebSocketConnectionD7_9.this._maxBinaryMessageSize;
int _maxTextMessage=WebSocketConnectionD10.this._maxTextMessageSize;
int _maxBinaryMessage=WebSocketConnectionD10.this._maxBinaryMessageSize;
/* ------------------------------------------------------------ */
public synchronized void sendMessage(String content) throws IOException
@ -347,7 +400,7 @@ public class WebSocketConnectionD7_9 extends AbstractConnection implements WebSo
if (_closedOut)
throw new IOException("closing");
byte[] data = content.getBytes(StringUtil.__UTF8);
_outbound.addFrame((byte)0x8,WebSocketConnectionD7_9.OP_TEXT,data,0,data.length);
_outbound.addFrame((byte)0x8,WebSocketConnectionD10.OP_TEXT,data,0,data.length);
checkWriteable();
_idle.access(_endp);
}
@ -357,7 +410,7 @@ public class WebSocketConnectionD7_9 extends AbstractConnection implements WebSo
{
if (_closedOut)
throw new IOException("closing");
_outbound.addFrame((byte)0x8,WebSocketConnectionD7_9.OP_BINARY,content,offset,length);
_outbound.addFrame((byte)0x8,WebSocketConnectionD10.OP_BINARY,content,offset,length);
checkWriteable();
_idle.access(_endp);
}
@ -400,7 +453,7 @@ public class WebSocketConnectionD7_9 extends AbstractConnection implements WebSo
if (_disconnecting)
return;
_disconnecting=true;
WebSocketConnectionD7_9.this.closeOut(code,message);
WebSocketConnectionD10.this.closeOut(code,message);
}
/* ------------------------------------------------------------ */
@ -525,7 +578,7 @@ public class WebSocketConnectionD7_9 extends AbstractConnection implements WebSo
{
boolean lastFrame = isLastFrame(flags);
synchronized(WebSocketConnectionD7_9.this)
synchronized(WebSocketConnectionD10.this)
{
// Ignore incoming after a close
if (_closedIn)
@ -550,10 +603,10 @@ public class WebSocketConnectionD7_9 extends AbstractConnection implements WebSo
switch(opcode)
{
case WebSocketConnectionD7_9.OP_CONTINUATION:
case WebSocketConnectionD10.OP_CONTINUATION:
{
// If text, append to the message buffer
if (_opcode==WebSocketConnectionD7_9.OP_TEXT && _connection.getMaxTextMessageSize()>=0)
if (_opcode==WebSocketConnectionD10.OP_TEXT && _connection.getMaxTextMessageSize()>=0)
{
if (_utf8.append(buffer.array(),buffer.getIndex(),buffer.length(),_connection.getMaxTextMessageSize()))
{
@ -568,7 +621,7 @@ public class WebSocketConnectionD7_9 extends AbstractConnection implements WebSo
}
else
{
_connection.close(WebSocketConnectionD7_9.CLOSE_LARGE,"Text message size > "+_connection.getMaxTextMessageSize()+" chars");
_connection.close(WebSocketConnectionD10.CLOSE_LARGE,"Text message size > "+_connection.getMaxTextMessageSize()+" chars");
_utf8.reset();
_opcode=-1;
}
@ -577,7 +630,7 @@ public class WebSocketConnectionD7_9 extends AbstractConnection implements WebSo
{
if (_aggregate.space()<_aggregate.length())
{
_connection.close(WebSocketConnectionD7_9.CLOSE_LARGE,"Message size > "+_connection.getMaxBinaryMessageSize());
_connection.close(WebSocketConnectionD10.CLOSE_LARGE,"Message size > "+_connection.getMaxBinaryMessageSize());
_aggregate.clear();
_opcode=-1;
}
@ -602,27 +655,27 @@ public class WebSocketConnectionD7_9 extends AbstractConnection implements WebSo
}
break;
}
case WebSocketConnectionD7_9.OP_PING:
case WebSocketConnectionD10.OP_PING:
{
Log.debug("PING {}",this);
if (!_closedOut)
_connection.sendControl(WebSocketConnectionD7_9.OP_PONG,buffer.array(),buffer.getIndex(),buffer.length());
_connection.sendControl(WebSocketConnectionD10.OP_PONG,buffer.array(),buffer.getIndex(),buffer.length());
break;
}
case WebSocketConnectionD7_9.OP_PONG:
case WebSocketConnectionD10.OP_PONG:
{
Log.debug("PONG {}",this);
break;
}
case WebSocketConnectionD7_9.OP_CLOSE:
case WebSocketConnectionD10.OP_CLOSE:
{
int code=-1;
int code=WebSocketConnectionD10.CLOSE_NOCODE;
String message=null;
if (buffer.length()>=2)
{
code=buffer.array()[buffer.getIndex()]*0xff+buffer.array()[buffer.getIndex()+1];
code=buffer.array()[buffer.getIndex()]*0x100+buffer.array()[buffer.getIndex()+1];
if (buffer.length()>2)
message=new String(buffer.array(),buffer.getIndex()+2,buffer.length()-2,StringUtil.__UTF8);
}
@ -631,7 +684,7 @@ public class WebSocketConnectionD7_9 extends AbstractConnection implements WebSo
}
case WebSocketConnectionD7_9.OP_TEXT:
case WebSocketConnectionD10.OP_TEXT:
{
if(_onTextMessage!=null)
{
@ -646,12 +699,12 @@ public class WebSocketConnectionD7_9 extends AbstractConnection implements WebSo
{
// If this is a text fragment, append to buffer
if (_utf8.append(buffer.array(),buffer.getIndex(),buffer.length(),_connection.getMaxTextMessageSize()))
_opcode=WebSocketConnectionD7_9.OP_TEXT;
_opcode=WebSocketConnectionD10.OP_TEXT;
else
{
_utf8.reset();
_opcode=-1;
_connection.close(WebSocketConnectionD7_9.CLOSE_LARGE,"Text message size > "+_connection.getMaxTextMessageSize()+" chars");
_connection.close(WebSocketConnectionD10.CLOSE_LARGE,"Text message size > "+_connection.getMaxTextMessageSize()+" chars");
}
}
}
@ -673,7 +726,7 @@ public class WebSocketConnectionD7_9 extends AbstractConnection implements WebSo
{
if (buffer.length()>_connection.getMaxBinaryMessageSize())
{
_connection.close(WebSocketConnectionD7_9.CLOSE_LARGE,"Message size > "+_connection.getMaxBinaryMessageSize());
_connection.close(WebSocketConnectionD10.CLOSE_LARGE,"Message size > "+_connection.getMaxBinaryMessageSize());
if (_aggregate!=null)
_aggregate.clear();
_opcode=-1;
@ -711,7 +764,7 @@ public class WebSocketConnectionD7_9 extends AbstractConnection implements WebSo
public String toString()
{
return WebSocketConnectionD7_9.this.toString()+"FH";
return WebSocketConnectionD10.this.toString()+"FH";
}
}

View File

@ -166,8 +166,9 @@ public class WebSocketFactory
case 7:
case 8:
case 9:
extensions= initExtensions(extensions_requested,8-WebSocketConnectionD7_9.OP_EXT_DATA, 16-WebSocketConnectionD7_9.OP_EXT_CTRL,3);
connection = new WebSocketConnectionD7_9(websocket, endp, _buffers, http.getTimeStamp(), _maxIdleTime, protocol,extensions,draft);
case 10:
extensions= initExtensions(extensions_requested,8-WebSocketConnectionD10.OP_EXT_DATA, 16-WebSocketConnectionD10.OP_EXT_CTRL,3);
connection = new WebSocketConnectionD10(websocket, endp, _buffers, http.getTimeStamp(), _maxIdleTime, protocol,extensions,draft);
break;
default:
Log.warn("Unsupported Websocket version: "+draft);

View File

@ -30,7 +30,7 @@ import org.eclipse.jetty.util.TypeUtil;
* threads will call the addMessage methods while other
* threads are flushing the generator.
*/
public class WebSocketGeneratorD7_9 implements WebSocketGenerator
public class WebSocketGeneratorD10 implements WebSocketGenerator
{
final private WebSocketBuffers _buffers;
final private EndPoint _endp;
@ -80,7 +80,7 @@ public class WebSocketGeneratorD7_9 implements WebSocketGenerator
final Random _random;
public RandomMaskGen()
{
_random=new SecureRandom();
_random=new Random();
}
public RandomMaskGen(Random random)
@ -95,14 +95,14 @@ public class WebSocketGeneratorD7_9 implements WebSocketGenerator
}
public WebSocketGeneratorD7_9(WebSocketBuffers buffers, EndPoint endp)
public WebSocketGeneratorD10(WebSocketBuffers buffers, EndPoint endp)
{
_buffers=buffers;
_endp=endp;
_maskGen=null;
}
public WebSocketGeneratorD7_9(WebSocketBuffers buffers, EndPoint endp, MaskGen maskGen)
public WebSocketGeneratorD10(WebSocketBuffers buffers, EndPoint endp, MaskGen maskGen)
{
_buffers=buffers;
_endp=endp;
@ -118,14 +118,14 @@ public class WebSocketGeneratorD7_9 implements WebSocketGenerator
if (_buffer==null)
_buffer=mask?_buffers.getBuffer():_buffers.getDirectBuffer();
boolean last=WebSocketConnectionD7_9.isLastFrame(flags);
boolean last=WebSocketConnectionD10.isLastFrame(flags);
byte orig=opcode;
int space=mask?14:10;
do
{
opcode = _opsent?WebSocketConnectionD7_9.OP_CONTINUATION:opcode;
opcode = _opsent?WebSocketConnectionD10.OP_CONTINUATION:opcode;
opcode=(byte)(((0xf&flags)<<4)+(0xf&opcode));
_opsent=true;

View File

@ -29,7 +29,7 @@ import org.eclipse.jetty.util.log.Log;
* Parser the WebSocket protocol.
*
*/
public class WebSocketParserD7_9 implements WebSocketParser
public class WebSocketParserD10 implements WebSocketParser
{
public enum State {
@ -72,7 +72,7 @@ public class WebSocketParserD7_9 implements WebSocketParser
* @param endp
* @param handler
*/
public WebSocketParserD7_9(WebSocketBuffers buffers, EndPoint endp, FrameHandler handler, boolean shouldBeMasked)
public WebSocketParserD10(WebSocketBuffers buffers, EndPoint endp, FrameHandler handler, boolean shouldBeMasked)
{
_buffers=buffers;
_endp=endp;
@ -160,11 +160,11 @@ public class WebSocketParserD7_9 implements WebSocketParser
_opcode=(byte)(b&0xf);
_flags=(byte)(0xf&(b>>4));
if (WebSocketConnectionD7_9.isControlFrame(_opcode)&&!WebSocketConnectionD7_9.isLastFrame(_flags))
if (WebSocketConnectionD10.isControlFrame(_opcode)&&!WebSocketConnectionD10.isLastFrame(_flags))
{
events++;
Log.warn("Fragmented Control from "+_endp);
_handler.close(WebSocketConnectionD7_9.CLOSE_PROTOCOL,"Fragmented control");
_handler.close(WebSocketConnectionD10.CLOSE_PROTOCOL,"Fragmented control");
_skip=true;
}
@ -205,7 +205,7 @@ public class WebSocketParserD7_9 implements WebSocketParser
if (_length>_buffer.capacity())
{
events++;
_handler.close(WebSocketConnectionD7_9.CLOSE_LARGE,"frame size "+_length+">"+_buffer.capacity());
_handler.close(WebSocketConnectionD10.CLOSE_LARGE,"frame size "+_length+">"+_buffer.capacity());
_skip=true;
}
@ -224,7 +224,7 @@ public class WebSocketParserD7_9 implements WebSocketParser
if (_length>=_buffer.capacity())
{
events++;
_handler.close(WebSocketConnectionD7_9.CLOSE_LARGE,"frame size "+_length+">"+_buffer.capacity());
_handler.close(WebSocketConnectionD10.CLOSE_LARGE,"frame size "+_length+">"+_buffer.capacity());
_skip=true;
}
@ -267,7 +267,7 @@ public class WebSocketParserD7_9 implements WebSocketParser
_buffer.skip(_bytesNeeded);
_state=State.START;
events++;
_handler.close(WebSocketConnectionD7_9.CLOSE_PROTOCOL,"bad mask");
_handler.close(WebSocketConnectionD10.CLOSE_PROTOCOL,"bad mask");
}
else
{

View File

@ -0,0 +1,715 @@
package org.eclipse.jetty.websocket;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.URI;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Exchanger;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import org.eclipse.jetty.util.BlockingArrayQueue;
import org.eclipse.jetty.websocket.WebSocket.Connection;
import org.junit.Assert;
import org.junit.Test;
public class WebSocketClientTest
{
@Test
public void testBadURL() throws Exception
{
WebSocketClient client = new WebSocketClient();
client.start();
boolean bad=false;
final AtomicBoolean open = new AtomicBoolean();
try
{
client.open(new URI("http://localhost:8080"),new WebSocket()
{
public void onOpen(Connection connection)
{
open.set(true);
}
public void onError(String message, Throwable ex)
{
}
public void onClose(int closeCode, String message)
{}
});
Assert.fail();
}
catch(IllegalArgumentException e)
{
bad=true;
}
Assert.assertTrue(bad);
Assert.assertFalse(open.get());
}
@Test
public void testBlockingConnectionRefused() throws Exception
{
WebSocketClient client = new WebSocketClient();
client.start();
client.setBlockingConnect(true);
boolean bad=false;
final AtomicBoolean open = new AtomicBoolean();
try
{
client.open(new URI("ws://127.0.0.1:1"),new WebSocket()
{
public void onOpen(Connection connection)
{
open.set(true);
}
public void onError(String message, Throwable ex)
{
}
public void onClose(int closeCode, String message)
{}
});
Assert.fail();
}
catch(IOException e)
{
bad=true;
}
Assert.assertTrue(bad);
Assert.assertFalse(open.get());
}
@Test
public void testAsyncConnectionRefused() throws Exception
{
WebSocketClient client = new WebSocketClient();
client.setConnectTimeout(1000);
client.start();
client.setBlockingConnect(false);
boolean bad=false;
final AtomicBoolean open = new AtomicBoolean();
final AtomicReference<String> error = new AtomicReference<String>(null);
final AtomicInteger close = new AtomicInteger();
final CountDownLatch latch = new CountDownLatch(1);
try
{
client.open(new URI("ws://127.0.0.1:1"),new WebSocket()
{
public void onOpen(Connection connection)
{
open.set(true);
latch.countDown();
}
public void onError(String message, Throwable ex)
{
error.set(message);
latch.countDown();
}
public void onClose(int closeCode, String message)
{
close.set(closeCode);
latch.countDown();
}
});
}
catch(IOException e)
{
bad=true;
}
Assert.assertFalse(bad);
Assert.assertFalse(open.get());
Assert.assertTrue(latch.await(1,TimeUnit.SECONDS));
Assert.assertNotNull(error.get());
}
@Test
public void testBlockingConnectionNotAccepted() throws Exception
{
WebSocketClient client = new WebSocketClient();
client.setConnectTimeout(500);
client.setBlockingConnect(true);
client.start();
ServerSocket server = new ServerSocket();
server.bind(null);
int port = server.getLocalPort();
boolean bad=false;
final AtomicReference<String> error = new AtomicReference<String>(null);
final CountDownLatch latch = new CountDownLatch(1);
try
{
client.open(new URI("ws://127.0.0.1:"+port),new WebSocket()
{
public void onOpen(Connection connection)
{
latch.countDown();
}
public void onError(String message, Throwable ex)
{
error.set(message);
latch.countDown();
}
public void onClose(int closeCode, String message)
{
latch.countDown();
}
});
}
catch(IOException e)
{
e.printStackTrace();
bad=true;
}
Assert.assertTrue(latch.await(1,TimeUnit.SECONDS));
Assert.assertTrue(bad||error.get()!=null);
}
@Test
public void testAsyncConnectionNotAccepted() throws Exception
{
WebSocketClient client = new WebSocketClient();
client.setBlockingConnect(true);
client.setConnectTimeout(300);
client.start();
ServerSocket server = new ServerSocket();
server.bind(null);
int port = server.getLocalPort();
boolean bad=false;
final AtomicBoolean open = new AtomicBoolean();
final AtomicReference<String> error = new AtomicReference<String>(null);
final AtomicInteger close = new AtomicInteger();
final CountDownLatch latch = new CountDownLatch(1);
try
{
client.open(new URI("ws://127.0.0.1:"+port),new WebSocket()
{
public void onOpen(Connection connection)
{
open.set(true);
latch.countDown();
}
public void onError(String message, Throwable ex)
{
error.set(message);
latch.countDown();
}
public void onClose(int closeCode, String message)
{
close.set(closeCode);
latch.countDown();
}
});
}
catch(IOException e)
{
bad=true;
}
Assert.assertFalse(bad);
Assert.assertFalse(open.get());
Assert.assertTrue(latch.await(1,TimeUnit.SECONDS));
Assert.assertNotNull(error.get());
}
@Test
public void testBlockingConnectionTimeout() throws Exception
{
WebSocketClient client = new WebSocketClient();
client.setConnectTimeout(500);
client.setBlockingConnect(true);
client.start();
ServerSocket server = new ServerSocket();
server.bind(null);
int port = server.getLocalPort();
boolean bad=false;
final AtomicReference<String> error = new AtomicReference<String>(null);
final CountDownLatch latch = new CountDownLatch(1);
try
{
client.open(new URI("ws://127.0.0.1:"+port),new WebSocket()
{
public void onOpen(Connection connection)
{
latch.countDown();
}
public void onError(String message, Throwable ex)
{
error.set(message);
latch.countDown();
}
public void onClose(int closeCode, String message)
{
latch.countDown();
}
});
}
catch(IOException e)
{
e.printStackTrace();
bad=true;
}
Assert.assertNotNull(server.accept());
Assert.assertTrue(latch.await(1,TimeUnit.SECONDS));
Assert.assertTrue(bad||error.get()!=null);
}
@Test
public void testAsyncConnectionTimeout() throws Exception
{
WebSocketClient client = new WebSocketClient();
client.setBlockingConnect(true);
client.setConnectTimeout(300);
client.start();
ServerSocket server = new ServerSocket();
server.bind(null);
int port = server.getLocalPort();
boolean bad=false;
final AtomicBoolean open = new AtomicBoolean();
final AtomicReference<String> error = new AtomicReference<String>(null);
final AtomicInteger close = new AtomicInteger();
final CountDownLatch latch = new CountDownLatch(1);
try
{
client.open(new URI("ws://127.0.0.1:"+port),new WebSocket()
{
public void onOpen(Connection connection)
{
open.set(true);
latch.countDown();
}
public void onError(String message, Throwable ex)
{
error.set(message);
latch.countDown();
}
public void onClose(int closeCode, String message)
{
close.set(closeCode);
latch.countDown();
}
});
}
catch(IOException e)
{
bad=true;
}
Assert.assertNotNull(server.accept());
Assert.assertFalse(bad);
Assert.assertFalse(open.get());
Assert.assertTrue(latch.await(1,TimeUnit.SECONDS));
Assert.assertNotNull(error.get());
}
@Test
public void testBadHandshake() throws Exception
{
WebSocketClient client = new WebSocketClient();
client.setBlockingConnect(true);
client.setConnectTimeout(300);
client.start();
ServerSocket server = new ServerSocket();
server.bind(null);
int port = server.getLocalPort();
boolean bad=false;
final AtomicBoolean open = new AtomicBoolean();
final AtomicReference<String> error = new AtomicReference<String>(null);
final AtomicInteger close = new AtomicInteger();
final CountDownLatch latch = new CountDownLatch(1);
try
{
client.open(new URI("ws://127.0.0.1:"+port),new WebSocket()
{
public void onOpen(Connection connection)
{
open.set(true);
latch.countDown();
}
public void onError(String message, Throwable ex)
{
error.set(message);
latch.countDown();
}
public void onClose(int closeCode, String message)
{
close.set(closeCode);
latch.countDown();
}
});
}
catch(IOException e)
{
bad=true;
}
Socket connection = server.accept();
BufferedReader in = new BufferedReader(new InputStreamReader(connection.getInputStream()));
for (String line=in.readLine();line!=null;line=in.readLine())
{
// System.err.println(line);
if (line.length()==0)
break;
}
connection.getOutputStream().write("HTTP/1.1 404 NOT FOUND\r\n\r\n".getBytes());
Assert.assertFalse(bad);
Assert.assertFalse(open.get());
Assert.assertTrue(latch.await(1,TimeUnit.SECONDS));
Assert.assertNotNull(error.get());
}
@Test
public void testBadUpgrade() throws Exception
{
WebSocketClient client = new WebSocketClient();
client.setBlockingConnect(true);
client.setConnectTimeout(10000);
client.start();
ServerSocket server = new ServerSocket();
server.bind(null);
int port = server.getLocalPort();
boolean bad=false;
final AtomicBoolean open = new AtomicBoolean();
final AtomicReference<String> error = new AtomicReference<String>(null);
final AtomicInteger close = new AtomicInteger();
final CountDownLatch latch = new CountDownLatch(1);
try
{
client.open(new URI("ws://127.0.0.1:"+port),new WebSocket()
{
public void onOpen(Connection connection)
{
open.set(true);
latch.countDown();
}
public void onError(String message, Throwable ex)
{
error.set(message);
latch.countDown();
}
public void onClose(int closeCode, String message)
{
close.set(closeCode);
latch.countDown();
}
});
}
catch(IOException e)
{
bad=true;
}
Socket connection = server.accept();
BufferedReader in = new BufferedReader(new InputStreamReader(connection.getInputStream()));
for (String line=in.readLine();line!=null;line=in.readLine())
{
// System.err.println(line);
if (line.length()==0)
break;
}
connection.getOutputStream().write((
"HTTP/1.1 101 Upgrade\r\n" +
"Sec-WebSocket-Accept: rubbish\r\n" +
"\r\n").getBytes());
Assert.assertFalse(bad);
Assert.assertFalse(open.get());
Assert.assertTrue(latch.await(1,TimeUnit.SECONDS));
Assert.assertNotNull(error.get());
}
@Test
public void testUpgrade() throws Exception
{
WebSocketClient client = new WebSocketClient();
client.setBlockingConnect(true);
client.setConnectTimeout(10000);
client.start();
ServerSocket server = new ServerSocket();
server.bind(null);
int port = server.getLocalPort();
boolean bad=false;
final AtomicBoolean open = new AtomicBoolean();
final AtomicReference<String> error = new AtomicReference<String>(null);
final AtomicInteger close = new AtomicInteger();
final CountDownLatch latch = new CountDownLatch(1);
try
{
client.open(new URI("ws://127.0.0.1:"+port),new WebSocket()
{
public void onOpen(Connection connection)
{
open.set(true);
latch.countDown();
}
public void onError(String message, Throwable ex)
{
error.set(message);
latch.countDown();
}
public void onClose(int closeCode, String message)
{
close.set(closeCode);
latch.countDown();
}
});
}
catch(IOException e)
{
bad=true;
}
Assert.assertFalse(bad);
String key="not sent";
Socket connection = server.accept();
BufferedReader in = new BufferedReader(new InputStreamReader(connection.getInputStream()));
for (String line=in.readLine();line!=null;line=in.readLine())
{
if (line.length()==0)
break;
if (line.startsWith("Sec-WebSocket-Key:"))
key=line.substring(18).trim();
}
connection.getOutputStream().write((
"HTTP/1.1 101 Upgrade\r\n" +
"Sec-WebSocket-Accept: "+ WebSocketConnectionD10.hashKey(key) +"\r\n" +
"\r\n").getBytes());
Assert.assertTrue(latch.await(1,TimeUnit.SECONDS));
Assert.assertNull(error.get());
Assert.assertTrue(open.get());
}
@Test
public void testIdle() throws Exception
{
WebSocketClient client = new WebSocketClient();
client.setBlockingConnect(true);
client.setConnectTimeout(10000);
client.setMaxIdleTime(500);
client.start();
ServerSocket server = new ServerSocket();
server.bind(null);
int port = server.getLocalPort();
boolean bad=false;
final AtomicBoolean open = new AtomicBoolean();
final AtomicInteger close = new AtomicInteger();
final CountDownLatch latch = new CountDownLatch(2);
try
{
client.open(new URI("ws://127.0.0.1:"+port),new WebSocket()
{
public void onOpen(Connection connection)
{
open.set(true);
latch.countDown();
}
public void onError(String message, Throwable ex)
{
latch.countDown();
}
public void onClose(int closeCode, String message)
{
close.set(closeCode);
latch.countDown();
}
});
}
catch(IOException e)
{
bad=true;
}
Assert.assertFalse(bad);
String key="not sent";
Socket connection = server.accept();
BufferedReader in = new BufferedReader(new InputStreamReader(connection.getInputStream()));
for (String line=in.readLine();line!=null;line=in.readLine())
{
if (line.length()==0)
break;
if (line.startsWith("Sec-WebSocket-Key:"))
key=line.substring(18).trim();
}
connection.getOutputStream().write((
"HTTP/1.1 101 Upgrade\r\n" +
"Sec-WebSocket-Accept: "+ WebSocketConnectionD10.hashKey(key) +"\r\n" +
"\r\n").getBytes());
Assert.assertTrue(latch.await(10,TimeUnit.SECONDS));
Assert.assertTrue(open.get());
Assert.assertEquals(WebSocketConnectionD10.CLOSE_NORMAL,close.get());
}
@Test
public void testNotIdle() throws Exception
{
WebSocketClient client = new WebSocketClient();
client.setBlockingConnect(true);
client.setConnectTimeout(10000);
client.setMaxIdleTime(500);
client.start();
ServerSocket server = new ServerSocket();
server.bind(null);
int port = server.getLocalPort();
boolean bad=false;
final AtomicBoolean open = new AtomicBoolean();
final Exchanger<Integer> close = new Exchanger<Integer>();
final CountDownLatch latch = new CountDownLatch(1);
final AtomicReference<WebSocket.Connection> connection = new AtomicReference<WebSocket.Connection>();
final BlockingQueue<String> queue = new BlockingArrayQueue<String>();
try
{
client.open(new URI("ws://127.0.0.1:"+port),new WebSocket.OnTextMessage()
{
public void onOpen(Connection c)
{
open.set(true);
connection.set(c);
latch.countDown();
}
public void onError(String message, Throwable ex)
{
latch.countDown();
}
public void onClose(int closeCode, String message)
{
try
{
close.exchange(closeCode);
}
catch(InterruptedException ex)
{}
latch.countDown();
}
public void onMessage(String data)
{
queue.add(data);
}
});
}
catch(IOException e)
{
bad=true;
}
Assert.assertFalse(bad);
String key="not sent";
Socket socket = server.accept();
BufferedReader in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
for (String line=in.readLine();line!=null;line=in.readLine())
{
if (line.length()==0)
break;
if (line.startsWith("Sec-WebSocket-Key:"))
key=line.substring(18).trim();
}
socket.getOutputStream().write((
"HTTP/1.1 101 Upgrade\r\n" +
"Sec-WebSocket-Accept: "+ WebSocketConnectionD10.hashKey(key) +"\r\n" +
"\r\n").getBytes());
Assert.assertTrue(latch.await(10,TimeUnit.SECONDS));
Assert.assertTrue(open.get());
// Send some messages client to server
byte[] recv = new byte[1024];
int len=-1;
for (int i=0;i<10;i++)
{
Thread.sleep(250);
connection.get().sendMessage("Hello");
len=socket.getInputStream().read(recv,0,recv.length);
Assert.assertTrue(len>0);
}
// Send some messages server to client
byte[] send = new byte[] { (byte)0x81, (byte) 0x02, (byte)'H', (byte)'i'};
for (int i=0;i<10;i++)
{
Thread.sleep(250);
socket.getOutputStream().write(send,0,send.length);
socket.getOutputStream().flush();
Assert.assertEquals("Hi",queue.poll(1,TimeUnit.SECONDS));
}
socket.getOutputStream().write(new byte[]{(byte)0x88, (byte) 0x02, (byte)4, (byte)87 },0,4);
socket.getOutputStream().flush();
Assert.assertEquals(new Integer(1111),close.exchange(null,1,TimeUnit.SECONDS));
}
}

View File

@ -11,7 +11,7 @@ import org.junit.Test;
/**
* @version $Revision$ $Date$
*/
public class WebSocketGeneratorD7_9Test
public class WebSocketGeneratorD10Test
{
private ByteArrayBuffer _out;
private WebSocketGenerator _generator;
@ -20,7 +20,7 @@ public class WebSocketGeneratorD7_9Test
byte[] _mask = new byte[4];
int _m;
public WebSocketGeneratorD7_9.MaskGen _maskGen = new WebSocketGeneratorD7_9.FixedMaskGen(
public WebSocketGeneratorD10.MaskGen _maskGen = new WebSocketGeneratorD10.FixedMaskGen(
new byte[]{(byte)0x00,(byte)0x00,(byte)0x0f,(byte)0xff});
@Before
@ -42,7 +42,7 @@ public class WebSocketGeneratorD7_9Test
@Test
public void testOneString() throws Exception
{
_generator = new WebSocketGeneratorD7_9(_buffers, _endPoint,null);
_generator = new WebSocketGeneratorD10(_buffers, _endPoint,null);
byte[] data = "Hell\uFF4F W\uFF4Frld".getBytes(StringUtil.__UTF8);
_generator.addFrame((byte)0x8,(byte)0x04,data,0,data.length);
@ -69,7 +69,7 @@ public class WebSocketGeneratorD7_9Test
@Test
public void testOneBuffer() throws Exception
{
_generator = new WebSocketGeneratorD7_9(_buffers, _endPoint,null);
_generator = new WebSocketGeneratorD10(_buffers, _endPoint,null);
String string = "Hell\uFF4F W\uFF4Frld";
byte[] bytes=string.getBytes(StringUtil.__UTF8);
@ -97,7 +97,7 @@ public class WebSocketGeneratorD7_9Test
@Test
public void testOneLongBuffer() throws Exception
{
_generator = new WebSocketGeneratorD7_9(_buffers, _endPoint,null);
_generator = new WebSocketGeneratorD10(_buffers, _endPoint,null);
byte[] b=new byte[150];
for (int i=0;i<b.length;i++)
@ -118,7 +118,7 @@ public class WebSocketGeneratorD7_9Test
@Test
public void testOneStringMasked() throws Exception
{
_generator = new WebSocketGeneratorD7_9(_buffers, _endPoint,_maskGen);
_generator = new WebSocketGeneratorD10(_buffers, _endPoint,_maskGen);
byte[] data = "Hell\uFF4F W\uFF4Frld".getBytes(StringUtil.__UTF8);
_generator.addFrame((byte)0x8,(byte)0x04,data,0,data.length);
@ -147,7 +147,7 @@ public class WebSocketGeneratorD7_9Test
@Test
public void testOneBufferMasked() throws Exception
{
_generator = new WebSocketGeneratorD7_9(_buffers, _endPoint,_maskGen);
_generator = new WebSocketGeneratorD10(_buffers, _endPoint,_maskGen);
String string = "Hell\uFF4F W\uFF4Frld";
byte[] bytes=string.getBytes(StringUtil.__UTF8);
@ -177,7 +177,7 @@ public class WebSocketGeneratorD7_9Test
@Test
public void testOneLongBufferMasked() throws Exception
{
_generator = new WebSocketGeneratorD7_9(_buffers, _endPoint,_maskGen);
_generator = new WebSocketGeneratorD10(_buffers, _endPoint,_maskGen);
byte[] b=new byte[150];
for (int i=0;i<b.length;i++)

View File

@ -25,6 +25,7 @@ import org.eclipse.jetty.server.Server;
import org.eclipse.jetty.server.handler.DefaultHandler;
import org.eclipse.jetty.server.nio.SelectChannelConnector;
import org.eclipse.jetty.util.StringUtil;
import org.eclipse.jetty.util.log.Log;
import org.eclipse.jetty.util.thread.QueuedThreadPool;
import org.junit.AfterClass;
import org.junit.BeforeClass;
@ -33,7 +34,7 @@ import org.junit.Test;
/**
* @version $Revision$ $Date$
*/
public class WebSocketLoadD7_9Test
public class WebSocketLoadD10Test
{
private static Server _server;
private static Connector _connector;
@ -116,6 +117,10 @@ public class WebSocketLoadD7_9Test
this.outbound = outbound;
}
public void onError(String message,Throwable ex)
{
}
public void onMessage(String data)
{
try
@ -142,8 +147,8 @@ public class WebSocketLoadD7_9Test
private final int iterations;
private final CountDownLatch latch;
private final SocketEndPoint _endp;
private final WebSocketGeneratorD7_9 _generator;
private final WebSocketParserD7_9 _parser;
private final WebSocketGeneratorD10 _generator;
private final WebSocketParserD10 _parser;
private final WebSocketParser.FrameHandler _handler = new WebSocketParser.FrameHandler()
{
public void onFrame(byte flags, byte opcode, Buffer buffer)
@ -167,8 +172,8 @@ public class WebSocketLoadD7_9Test
this.iterations = iterations;
_endp=new SocketEndPoint(socket);
_generator = new WebSocketGeneratorD7_9(new WebSocketBuffers(32*1024),_endp,new WebSocketGeneratorD7_9.FixedMaskGen());
_parser = new WebSocketParserD7_9(new WebSocketBuffers(32*1024),_endp,_handler,false);
_generator = new WebSocketGeneratorD10(new WebSocketBuffers(32*1024),_endp,new WebSocketGeneratorD10.FixedMaskGen());
_parser = new WebSocketParserD10(new WebSocketBuffers(32*1024),_endp,_handler,false);
}
@ -202,7 +207,7 @@ public class WebSocketLoadD7_9Test
for (int i = 0; i < iterations; ++i)
{
byte[] data = message.getBytes(StringUtil.__UTF8);
_generator.addFrame((byte)0x8,WebSocketConnectionD7_9.OP_TEXT,data,0,data.length);
_generator.addFrame((byte)0x8,WebSocketConnectionD10.OP_TEXT,data,0,data.length);
_generator.flush();
//System.err.println("-> "+message);

View File

@ -250,6 +250,10 @@ public class WebSocketMessageD00Test
return latch.await(time, TimeUnit.MILLISECONDS);
}
public void onError(String message,Throwable ex)
{
}
public void onClose(int code,String message)
{
}

View File

@ -756,6 +756,10 @@ public class WebSocketMessageD06Test
{
this.connection = connection;
}
public void onError(String message,Throwable ex)
{
}
public void onOpen(Connection connection)
{

View File

@ -34,7 +34,7 @@ import org.junit.Test;
/**
* @version $Revision$ $Date$
*/
public class WebSocketMessageD7_9Test
public class WebSocketMessageD10Test
{
private static Server _server;
private static Connector _connector;
@ -75,7 +75,7 @@ public class WebSocketMessageD7_9Test
@Test
public void testHash()
{
assertEquals("s3pPLMBiTxaQ9kYGzzhZRbK+xOo=",WebSocketConnectionD7_9.hashKey("dGhlIHNhbXBsZSBub25jZQ=="));
assertEquals("s3pPLMBiTxaQ9kYGzzhZRbK+xOo=",WebSocketConnectionD10.hashKey("dGhlIHNhbXBsZSBub25jZQ=="));
}
@Test
@ -116,7 +116,7 @@ public class WebSocketMessageD7_9Test
String data=message.toString();
_serverWebSocket.connection.sendMessage(data);
assertEquals(WebSocketConnectionD7_9.OP_TEXT,input.read());
assertEquals(WebSocketConnectionD10.OP_TEXT,input.read());
assertEquals(0x7e,input.read());
assertEquals(0x1f,input.read());
assertEquals(0xf6,input.read());
@ -322,7 +322,7 @@ public class WebSocketMessageD7_9Test
output.write(buf,0,l+3);
output.flush();
assertEquals(0x40+WebSocketConnectionD7_9.OP_TEXT,input.read());
assertEquals(0x40+WebSocketConnectionD10.OP_TEXT,input.read());
assertEquals(0x20+3,input.read());
assertEquals(0x7e,input.read());
assertEquals(0x02,input.read());
@ -490,7 +490,7 @@ public class WebSocketMessageD7_9Test
output.write(bytes[i]^0xff);
output.flush();
assertEquals(0x80|WebSocketConnectionD7_9.OP_CLOSE,input.read());
assertEquals(0x80|WebSocketConnectionD10.OP_CLOSE,input.read());
assertEquals(30,input.read());
int code=(0xff&input.read())*0x100+(0xff&input.read());
assertEquals(1004,code);
@ -541,7 +541,7 @@ public class WebSocketMessageD7_9Test
assertEquals(0x80|WebSocketConnectionD7_9.OP_CLOSE,input.read());
assertEquals(0x80|WebSocketConnectionD10.OP_CLOSE,input.read());
assertEquals(30,input.read());
int code=(0xff&input.read())*0x100+(0xff&input.read());
assertEquals(1004,code);
@ -577,7 +577,7 @@ public class WebSocketMessageD7_9Test
assertNotNull(_serverWebSocket.connection);
_serverWebSocket.getConnection().setMaxBinaryMessageSize(1024);
output.write(WebSocketConnectionD7_9.OP_BINARY);
output.write(WebSocketConnectionD10.OP_BINARY);
output.write(0x8a);
output.write(0xff);
output.write(0xff);
@ -598,7 +598,7 @@ public class WebSocketMessageD7_9Test
output.write(bytes[i]^0xff);
output.flush();
assertEquals(0x80+WebSocketConnectionD7_9.OP_BINARY,input.read());
assertEquals(0x80+WebSocketConnectionD10.OP_BINARY,input.read());
assertEquals(20,input.read());
lookFor("01234567890123456789",input);
}
@ -655,7 +655,7 @@ public class WebSocketMessageD7_9Test
output.flush();
assertEquals(0x80|WebSocketConnectionD7_9.OP_CLOSE,input.read());
assertEquals(0x80|WebSocketConnectionD10.OP_CLOSE,input.read());
assertEquals(19,input.read());
int code=(0xff&input.read())*0x100+(0xff&input.read());
assertEquals(1004,code);
@ -704,7 +704,7 @@ public class WebSocketMessageD7_9Test
output.write(bytes[i]^0xff);
output.flush();
assertEquals(0x80|WebSocketConnectionD7_9.OP_CLOSE,input.read());
assertEquals(0x80|WebSocketConnectionD10.OP_CLOSE,input.read());
assertEquals(19,input.read());
int code=(0xff&input.read())*0x100+(0xff&input.read());
assertEquals(1004,code);
@ -830,14 +830,14 @@ public class WebSocketMessageD7_9Test
final AtomicReference<String> received = new AtomicReference<String>();
ByteArrayEndPoint endp = new ByteArrayEndPoint(new byte[0],4096);
WebSocketGeneratorD7_9 gen = new WebSocketGeneratorD7_9(new WebSocketBuffers(8096),endp,null);
WebSocketGeneratorD10 gen = new WebSocketGeneratorD10(new WebSocketBuffers(8096),endp,null);
byte[] data = message.getBytes(StringUtil.__UTF8);
gen.addFrame((byte)0x8,(byte)0x4,data,0,data.length);
endp = new ByteArrayEndPoint(endp.getOut().asArray(),4096);
WebSocketParserD7_9 parser = new WebSocketParserD7_9(new WebSocketBuffers(8096),endp,new WebSocketParser.FrameHandler()
WebSocketParserD10 parser = new WebSocketParserD10(new WebSocketBuffers(8096),endp,new WebSocketParser.FrameHandler()
{
public void onFrame(byte flags, byte opcode, Buffer buffer)
{
@ -862,15 +862,15 @@ public class WebSocketMessageD7_9Test
final AtomicReference<String> received = new AtomicReference<String>();
ByteArrayEndPoint endp = new ByteArrayEndPoint(new byte[0],4096);
WebSocketGeneratorD7_9.MaskGen maskGen = new WebSocketGeneratorD7_9.RandomMaskGen();
WebSocketGeneratorD10.MaskGen maskGen = new WebSocketGeneratorD10.RandomMaskGen();
WebSocketGeneratorD7_9 gen = new WebSocketGeneratorD7_9(new WebSocketBuffers(8096),endp,maskGen);
WebSocketGeneratorD10 gen = new WebSocketGeneratorD10(new WebSocketBuffers(8096),endp,maskGen);
byte[] data = message.getBytes(StringUtil.__UTF8);
gen.addFrame((byte)0x8,(byte)0x1,data,0,data.length);
endp = new ByteArrayEndPoint(endp.getOut().asArray(),4096);
WebSocketParserD7_9 parser = new WebSocketParserD7_9(new WebSocketBuffers(8096),endp,new WebSocketParser.FrameHandler()
WebSocketParserD10 parser = new WebSocketParserD10(new WebSocketBuffers(8096),endp,new WebSocketParser.FrameHandler()
{
public void onFrame(byte flags, byte opcode, Buffer buffer)
{
@ -982,6 +982,11 @@ public class WebSocketMessageD7_9Test
return disconnected.await(time, TimeUnit.MILLISECONDS);
}
public void onError(String message,Throwable ex)
{
disconnected.countDown();
}
public void onClose(int code,String message)
{
disconnected.countDown();
@ -993,9 +998,9 @@ public class WebSocketMessageD7_9Test
{
switch(opcode)
{
case WebSocketConnectionD7_9.OP_CLOSE:
case WebSocketConnectionD7_9.OP_PING:
case WebSocketConnectionD7_9.OP_PONG:
case WebSocketConnectionD10.OP_CLOSE:
case WebSocketConnectionD10.OP_PING:
case WebSocketConnectionD10.OP_PONG:
break;
default:

View File

@ -20,7 +20,7 @@ import org.junit.Test;
/**
* @version $Revision$ $Date$
*/
public class WebSocketParserD7_9Test
public class WebSocketParserD10Test
{
private MaskedByteArrayBuffer _in;
private Handler _handler;
@ -87,7 +87,7 @@ public class WebSocketParserD7_9Test
ByteArrayEndPoint endPoint = new ByteArrayEndPoint();
endPoint.setNonBlocking(true);
_handler = new Handler();
_parser=new WebSocketParserD7_9(buffers, endPoint,_handler,true);
_parser=new WebSocketParserD10(buffers, endPoint,_handler,true);
_in = new MaskedByteArrayBuffer();
endPoint.setIn(_in);
@ -187,7 +187,7 @@ public class WebSocketParserD7_9Test
{
WebSocketBuffers buffers = new WebSocketBuffers(0x20000);
ByteArrayEndPoint endPoint = new ByteArrayEndPoint();
WebSocketParser parser=new WebSocketParserD7_9(buffers, endPoint,_handler,false);
WebSocketParser parser=new WebSocketParserD10(buffers, endPoint,_handler,false);
ByteArrayBuffer in = new ByteArrayBuffer(0x20000);
endPoint.setIn(in);
@ -261,7 +261,7 @@ public class WebSocketParserD7_9Test
assertTrue(progress>0);
assertEquals(WebSocketConnectionD7_9.CLOSE_LARGE,_handler._code);
assertEquals(WebSocketConnectionD10.CLOSE_LARGE,_handler._code);
for (int i=0;i<2048;i++)
_in.put((byte)'a');
progress =_parser.parseNext();

View File

@ -321,6 +321,7 @@
<module>jetty-nested</module>
<module>jetty-overlay-deployer</module>
<module>jetty-osgi</module>
<module>jetty-nosql</module>
<module>test-continuation</module>
<module>test-continuation-jetty6</module>
<module>test-jetty-servlet</module>

View File

@ -68,6 +68,12 @@ public class WebSocketChatServlet extends WebSocketServlet
}
}
public void onError(String message,Throwable ex)
{
Log.warn(this+" onError",ex);
_members.remove(this);
}
public void onClose(int code, String message)
{
// Log.info(this+" onDisconnect");