Получение транспорта и исключение EOF с CometD на стороне клиента
Я получаю исключение транспорта и EOF на стороне клиента при использовании CometD. Я пытался найти решение с помощью документации по CometD, а также гуглил, но не смог найти никакого решения.
Ниже приведена трассировка стека исключений, которые я получаю
CommetD Connected!!!!!!!
cometDURL http://192.168.210.106:8080/cometd2/cometd
subscriberChannel /service/java
In finally...........
2013-09-12 10:20:26,200 [HttpClient-201] INFO org.cometd.client.BayeuxClient.180910783 - Messages failed [{id=31, connectionType=long-polling, channel=/meta/connect, clientId=2qpitkhfcjh1g511al4xal9vltg}]
org.cometd.common.TransportException at org.cometd.client.transport.LongPollingTransport$TransportExchange.onResponseComplete(LongPollingTransport.java:334) at org.eclipse.jetty.client.HttpExchange$Listener.onResponseComplete(HttpExchange.java:999)
at org.eclipse.jetty.client.HttpExchange.setStatus(HttpExchange.java:288)
at org.eclipse.jetty.client.HttpConnection$Handler.messageComplete(HttpConnection.java:628)
at org.eclipse.jetty.http.HttpParser.parseNext(HttpParser.java:786)
at org.eclipse.jetty.http.HttpParser.parseAvailable(HttpParser.java:220)
at org.eclipse.jetty.client.HttpConnection.handle(HttpConnection.java:275)
at org.eclipse.jetty.io.nio.SelectChannelEndPoint.handle(SelectChannelEndPoint.java:545)
at org.eclipse.jetty.io.nio.SelectChannelEndPoint$1.run(SelectChannelEndPoint.java:43)
at org.eclipse.jetty.util.thread.QueuedThreadPool$3.run(QueuedThreadPool.java:529)
at java.lang.Thread.run(Thread.java:662)
java.io.EOFException: closed: TransportExchange@50115a3f=POST//192.168.145.210:8080/cometd2/cometd#WAITING(64ms)sent=64ms
at org.eclipse.jetty.client.HttpConnection.close(HttpConnection.java:663)
at org.eclipse.jetty.client.HttpDestination.close(HttpDestination.java:639)
at org.eclipse.jetty.client.HttpClient.doStop(HttpClient.java:491)
at org.eclipse.jetty.util.component.AbstractLifeCycle.stop(AbstractLifeCycle.java:83)
at com.commetd.CommetD.publish(CommetD.java:58)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
at java.lang.reflect.Method.invoke(Method.java:597)
at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:301)
at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:182)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:149)
at org.springframework.transaction.interceptor.TransactionInterceptor.invoke(TransactionInterceptor.java:106)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:171)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:171)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:171)
at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:204)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
at java.lang.reflect.Method.invoke(Method.java:597)
at sun.reflect.GeneratedMethodAccessor72.invoke(Unknown Source)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
at java.lang.reflect.Method.invoke(Method.java:597)
at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:301)
at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:182)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:149)
at org.springframework.transaction.interceptor.TransactionInterceptor.invoke(TransactionInterceptor.java:106)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:171)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:171)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:171)
at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:204)
at $Proxy25.callProxy(Unknown Source)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
at java.lang.reflect.Method.invoke(Method.java:597)
at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:301)
at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:182)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:149)
at org.springframework.transaction.interceptor.TransactionInterceptor.invoke(TransactionInterceptor.java:106)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:171)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:171)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:171)
at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:204)
at $Proxy171.publishEventToBulletinManager(Unknown Source)
at java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:908)
at java.lang.Thread.run(Thread.java:662)
Я использовал следующий код для подключения к серверу CometD
public static boolean publish(String cometDURL,String subscriberChannel,Map<String,Object> msgToPublish) throws Exception{
boolean isSuccess = true;
HttpClient httpClient = new HttpClient();
BayeuxClient client = null;
ClientTransport transport=null;
// Here setup Jetty's HttpClient, for example:
httpClient.setMaxConnectionsPerAddress(10);
try {
httpClient.start();
Map<String, Object> options = new HashMap<String, Object>();
transport = LongPollingTransport.create(options, httpClient);
client = new BayeuxClient(cometDURL, transport);
client.handshake();
boolean handshaken = client.waitFor(1000, BayeuxClient.State.CONNECTED);
if(handshaken){
System.out.println("CommetD Connected!!!!!!!");
/*Map<String,Object> data = new HashMap<String,Object>();
data.put("name", "Java msg......... Dynamic msg.....");*/
// Fill in the data
client.getChannel(subscriberChannel).publish(msgToPublish);
}else{
isSuccess=false;
System.out.println("CommetD not connected:-(");
}
} catch (Exception e) {
isSuccess=false;
// TODO Auto-generated catch block
e.printStackTrace();
}finally{
client.disconnect();
client.waitFor(1000, BayeuxClient.State.DISCONNECTED);
transport.terminate();
httpClient.stop();
}
return isSuccess;
}
Код на стороне сервера
@Listener("/service/java")
public void processMsgFromJava(ServerSession remote, ServerMessage.Mutable message)
{
System.out.println("Received msg from java..................");
Map<String, Object> input = message.getDataAsMap();
String eventId = (String)input.get("eventID");
String updatedDate = (String)input.get("updatedDate");
// String channelName =(String)input.get("ChannelToPublish");
System.out.println("msg received : "+eventId+":"+updatedDate);
// Map<String, Object> output = new HashMap<String, Object>();
//output.put("greeting", name);
// remote.deliver(serverSession, "/java/test", output, null);
String channelName = "/java/test";
// Initialize the channel, making it persistent and lazy
bayeux.createIfAbsent(channelName, new ConfigurableServerChannel.Initializer()
{
public void configureChannel(ConfigurableServerChannel channel)
{
channel.setPersistent(true);
channel.setLazy(true);
}
});
// System.out.println("HHHHHHHHHHHHREEEEEEEEEEEEEEEEEEEEEEEEERRRRRRRRRRRRRRRRRRRRRRRRRRRRREEEEEEEEEE");
// Convert the Update business object to a CometD-friendly format
/* Map<String, Object> data = new HashMap<String, Object>();
data.put("javamsg", name);*/
// Publish to all subscribers
ServerChannel channel = bayeux.getChannel(channelName);
channel.publish(serverSession, input, null);
}
JavaScript на стороне клиента:
(function($)
{
var cometd = $.cometd;
$(document).ready(function()
{
function _connectionEstablished()
{
$('#body').append('<div>CometD Connection Established</div>');
}
function _connectionBroken()
{
$('#body').append('<div>CometD Connection Broken</div>');
}
function _connectionClosed()
{
$('#body').append('<div>CometD Connection Closed</div>');
}
// Function that manages the connection status with the Bayeux server
var _connected = false;
function _metaConnect(message)
{
if (cometd.isDisconnected())
{
_connected = false;
_connectionClosed();
return;
}
var wasConnected = _connected;
_connected = message.successful === true;
if (!wasConnected && _connected)
{
_connectionEstablished();
}
else if (wasConnected && !_connected)
{
_connectionBroken();
}
}
// Function invoked when first contacting the server and
// when the server has lost the state of this client
function _metaHandshake(handshake)
{
if (handshake.successful === true)
{
cometd.batch(function()
{
cometd.subscribe('/java/test', function(message)
{
$('#body').append('<div>Server Says: ' + message.data.eventID + ':'+ message.data.updatedDate + '</div>');
});
// Publish on a service channel since the message is for the server only
// cometd.publish('/service/hello', { name: 'World' });
});
}
}
// Disconnect when the page unloads
$(window).unload(function()
{
cometd.disconnect(true);
});
var cometURL = location.protocol + "//" + location.host + config.contextPath + "/cometd";
cometd.configure({
url: cometURL,
logLevel: 'debug'
});
cometd.addListener('/meta/handshake', _metaHandshake);
cometd.addListener('/meta/connect', _metaConnect);
cometd.handshake();
});
})(jQuery);
Может ли кто-нибудь направить меня в решении этого вопроса. Я не уверен, где я иду не так.
1 ответ
Прежде всего, вы используете то, что называется com.commetd.CommetD это не стандартный код CometD. Не ясно, является ли код, который вы показываете, кодом для этого класса.
Предполагая, что это простая оболочка для исходного кода CometD, трассировка стека объясняет все это: вы останавливаете HttpClient изнутри publish() метод, который не имеет смысла.
Пожалуйста, внимательно прочитайте концепции CometD и как использовать клиентскую библиотеку CometD.