Commit b3e55f3a by amir

before testing server zmq

parent 6a10777c
...@@ -3,7 +3,13 @@ package microservice.services.protocol.zmq; ...@@ -3,7 +3,13 @@ package microservice.services.protocol.zmq;
import microservice.common.context.RestMsg; import microservice.common.context.RestMsg;
import microservice.io.iface.IRequest; import microservice.io.iface.IRequest;
import microservice.io.iface.IResponse; import microservice.io.iface.IResponse;
import microservice.services.IRestServiceZmqImpl;
import microservice.utils.ZSocketPool;
import org.apache.commons.lang.SerializationUtils;
import org.zeromq.ZMQ;
import org.zeromq.ZSocket;
import java.io.ByteArrayInputStream;
import java.io.InputStream; import java.io.InputStream;
import java.nio.ByteBuffer; import java.nio.ByteBuffer;
import java.util.List; import java.util.List;
...@@ -22,17 +28,17 @@ public class RestImpl { ...@@ -22,17 +28,17 @@ public class RestImpl {
@Override @Override
public InputStream getInputStream() { public InputStream getInputStream() {
return null; return new ByteArrayInputStream(restMsg.content().getBytes());
} }
@Override @Override
public String getQueryString() { public String getQueryString() {
return null; return restMsg.queryString();
} }
@Override @Override
public String getRelativePath() { public String getRelativePath() {
return null; return restMsg.url();
} }
@Override @Override
...@@ -47,25 +53,39 @@ public class RestImpl { ...@@ -47,25 +53,39 @@ public class RestImpl {
@Override @Override
public boolean startAsync(Runnable asyncFunc) { public boolean startAsync(Runnable asyncFunc) {
return false; return true;
} }
} }
public static class IResponseZmqRestImpl implements IResponse { public static class IResponseZmqRestImpl implements IResponse {
RestMsg restMsg = null; RestMsg restMsg = null;
public IResponseZmqRestImpl(RestMsg restMsg) { ZSocketPool serverSendPool = null;
this.restMsg = restMsg; long rcid = 0;
public IResponseZmqRestImpl(RestMsg restMsg, ZSocketPool serverSendPool, long rcid) {
this.restMsg = restMsg;
this.serverSendPool = serverSendPool;
this.rcid = rcid;
} }
@Override @Override
public void send(ByteBuffer buffer) { public void send(ByteBuffer buffer) {
send(new String(buffer.array()));
} }
@Override @Override
public void send(String response) { public void send(String response) {
ZSocket socket = null;
try {
byte[] data = SerializationUtils.serialize(new IRestServiceZmqImpl.ServerReplyMsg(rcid,restMsg.source(),response));
socket = serverSendPool.borrowObject();
socket.send(data, ZMQ.DONTWAIT);
} catch (Exception e) {
e.printStackTrace();
} finally {
if (socket != null)
serverSendPool.returnObject(socket);
}
} }
} }
} }
\ No newline at end of file
...@@ -5,6 +5,8 @@ import com.google.common.cache.CacheBuilder; ...@@ -5,6 +5,8 @@ import com.google.common.cache.CacheBuilder;
import io.undertow.predicate.Predicate; import io.undertow.predicate.Predicate;
import io.undertow.util.PathTemplateMatcher; import io.undertow.util.PathTemplateMatcher;
import microservice.services.CommonServices; import microservice.services.CommonServices;
import microservice.services.IRestServiceZmqImpl;
import org.apache.commons.lang.SerializationUtils;
import org.junit.Test; import org.junit.Test;
import java.util.Arrays; import java.util.Arrays;
...@@ -181,4 +183,5 @@ public class TestServicesAndMethods { ...@@ -181,4 +183,5 @@ public class TestServicesAndMethods {
System.out.println("Remove gcache Test of: " + String.valueOf(ITERATIONS) +" took (msec): " + String.valueOf(System.currentTimeMillis() - start)); System.out.println("Remove gcache Test of: " + String.valueOf(ITERATIONS) +" took (msec): " + String.valueOf(System.currentTimeMillis() - start));
} }
} }
...@@ -6,6 +6,8 @@ import itc.ItcMessageQueue; ...@@ -6,6 +6,8 @@ import itc.ItcMessageQueue;
import microservice.common.context.CrudMethod; import microservice.common.context.CrudMethod;
import microservice.common.context.RestMsg; import microservice.common.context.RestMsg;
import microservice.common.context.RestResponse; import microservice.common.context.RestResponse;
import microservice.services.IRestServiceZmqImpl;
import org.apache.commons.lang.SerializationUtils;
import org.junit.Test; import org.junit.Test;
import org.zeromq.*; import org.zeromq.*;
import rx.Observable; import rx.Observable;
...@@ -406,4 +408,13 @@ public class TestZMQ { ...@@ -406,4 +408,13 @@ public class TestZMQ {
} }
@Test
public void testSerialize(){
byte[] data = SerializationUtils.serialize(new IRestServiceZmqImpl.ServerReplyMsg(2,SOURCE_CHANNEL,JSON_CONTENT));
IRestServiceZmqImpl.ServerReplyMsg serverReplyMsg = (IRestServiceZmqImpl.ServerReplyMsg) SerializationUtils.deserialize(data);
System.out.println(serverReplyMsg.getReplyAddress());
}
} }
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or sign in to comment