在GWT中使用Jetty的Comet技术

江湖浪人

web.xml中配置ContinuationFilter

<filter>
<filter-name>jetty</filter-name>
<filter-class>org.eclipse.jetty.continuation.ContinuationFilter</filter-class>
<init-param >
<param-name>faux</param-name>
<param-value>true</param-value>
</init-param>
</filter>
<filter-mapping>
<filter-name>jetty</filter-name>
<servlet-name>comet</servlet-name>
</filter-mapping>


GWT Service中的设置

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
import javax.servlet.ServletConfig;
import javax.servlet.ServletException;
import javax.servlet.http.HttpServletRequest;

import net.sf.hibernate4gwt.core.HibernateBeanManager;
import net.sf.hibernate4gwt.gwt.HibernateRemoteService;
import org.eclipse.jetty.continuation.Continuation;
import org.eclipse.jetty.continuation.ContinuationThrowable;
import org.springframework.web.context.WebApplicationContext;
import org.springframework.web.context.support.WebApplicationContextUtils;
import com.extjs.gxt.ui.client.data.BaseModelData;
import com.extjs.gxt.ui.client.data.ModelData;
import com.google.gwt.user.server.rpc.UnexpectedException;
import com.single.desktopoa.client.event.CometEvents;
import com.single.desktopoa.common.person.Person;
import com.single.desktopoa.exception.RetryException;
public class BaseService extends HibernateRemoteService {
public static final String PAYLOAD = "com.google.gwt.payload";
public static final String PERSON = "person";
public static final String COMET_UTIL = "CometUtil";
public static final CometUtil cometUtil=new CometUtil();
protected Person person;
protected WebApplicationContext springContext;
@Override
public void init(ServletConfig config) throws ServletException {
super.init(config);
springContext = WebApplicationContextUtils.getWebApplicationContext(
config.getServletContext(),
WebApplicationContext.ROOT_WEB_APPLICATION_CONTEXT_ATTRIBUTE);
setBeanManager((HibernateBeanManager) springContext
.getBean("hibernateBeanManager"));
}
@Override
protected String readContent(HttpServletRequest request)
throws ServletException, IOException {
String payload = (String) request.getAttribute(PAYLOAD);
if (payload == null) {
payload = super.readContent(request);
request.setAttribute(PAYLOAD, payload);
}
return payload;
}
@Override
protected void doUnexpectedFailure(Throwable e) {
throwIfRetryRequest(e);
super.doUnexpectedFailure(e);
}
//如果我们的异常不想被GWT捕获，在该方法中向上抛出
protected void throwIfRetryRequest(Throwable caught) {
if (caught instanceof UnexpectedException) {
caught = caught.getCause();
}
//如果抛出的是自定义的，为了Hold request的异常，则继续往上抛出 ContinuationThrowable()
//此ContinuationThrowable() 是一个Error，ContinuationFilter会捕获这个Error，为了能让Filter捕获，我手动向上抛出，很笨拙的方法，期待以后改进
if(caught instanceof RetryException){
throw new ContinuationThrowable();
}
}
@Override
protected void onBeforeRequestDeserialized(String serializedRequest) {
person = (Person) getThreadLocalRequest().getSession().getAttribute(
PERSON);
super.onBeforeRequestDeserialized(serializedRequest);
}
public class CometMember {
public Person person;
public Continuation continuation;
public Queue<ModelData> queue = new LinkedList<ModelData>();
public void addEvent(String event,Object data) {
boolean isContain = false;
for (Iterator<ModelData> it = queue.iterator(); it.hasNext();) {
if (it.next().get("event").equals(event)) {
isContain = true;
break;
}
}
if (!isContain) {
ModelData modeldata=new BaseModelData();
modeldata.set("event", event);
if(data!=null)
modeldata.set("data", data);
queue.add(modeldata);
}
synchronized (this) {
//if (continuation != null) {
//continuation.resume();
//}
}
}
}
public static class CometUtil {
protected HashMap<Long, CometMember> memberMap = new HashMap<Long, CometMember>();
public List<Person> getOnlineList(){
List<Person> onlines=new ArrayList<Person>();
for(CometMember member:memberMap.values()){
onlines.add(member.person);
}
return onlines;
}
public boolean addEvent(Person receiver, String event,Object data) {
if (isContain(receiver)) {
CometMember member = memberMap.get(receiver.getId());
member.addEvent(event,data);
return true;
}
return false;
}
public void addMember(CometMember member) {
for(Iterator<CometMember> it=memberMap.values().iterator();it.hasNext();){
CometMember mem=it.next();
mem.addEvent(CometEvents.WORKTALK_newlogin, null);
}
memberMap.put(member.person.getId(), member);
//test
member.addEvent(CometEvents.WORKTALK_newlogin, null);
}
private boolean isContain(Person receiver) {
if (memberMap.get(receiver.getId()) == null) {
return false;
}
return true;
}
}
}


处理Comet请求

import java.util.ArrayList;
import java.util.List;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import org.eclipse.jetty.continuation.Continuation;
import org.eclipse.jetty.continuation.ContinuationSupport;
import org.mortbay.jetty.RetryRequest;
import com.extjs.gxt.ui.client.data.BaseModelData;
import com.extjs.gxt.ui.client.data.ModelData;
import com.single.desktopoa.exception.RetryException;
import com.single.desktopoa.module.comet.CometService;
import com.single.desktopoa.server.BaseService;
public class CometServiceImpl extends BaseService implements CometService {



public List<ModelData> getEvents() {
HttpServletRequest request=getThreadLocalRequest();
HttpServletResponse response=getThreadLocalResponse();

CometMember member=(CometMember)request.getSession().getAttribute(MEMBER);
if(member==null){
member=new CometMember();
member.person=person;
request.getSession().setAttribute(MEMBER, member);
cometUtil.addMember(member);
}
if(member.queue.size()>0){
List<ModelData> result=new ArrayList<ModelData>();
while(member.queue.size()>0){
result.add(member.queue.poll());
}
return result;
}else {
Continuation continuation=ContinuationSupport.getContinuation(request);
if(continuation.isInitial()){
//如果是新创建的
continuation.suspend(response);
member.continuation=continuation;
}else{
return new ArrayList<ModelData>();
}
}
//手动抛出异常，表明此请求需要Hold住，待以后唤醒
throw new RetryException();

}

}