flex mina2.0 AMF3实例(转) -

istudy

浏览: 24710 次
性别:
来自: 杭州

最近访客更多访客>>

devcang

博主相关

博客

微博

相册

留言

关于我

文章分类

全部博客 (14)

社区版块

存档分类

flex mina2.0 AMF3实例(转)

Flex Mina Java Apache Socket

/**
* 类功能描述:通讯服务器
* 类创建人:
* 创建时间：
*/
import java.io.IOException;
import java.net.InetSocketAddress;
import org.apache.mina.core.service.IoAcceptor;
import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.logging.LoggingFilter;
import org.apache.mina.transport.socket.nio.NioSocketAcceptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import common.AMF3CodecFactory;
public class ServerMain {
private static final int socketPORT = 10012;
public ServerMain(){
//创建服务端连接
final IoAcceptor acceptor = new NioSocketAcceptor();
//创建日志过滤器进行日志处理
acceptor.getFilterChain().addLast(“logger”, new LoggingFilter());
//创建AMF3编解码过滤器编解码
acceptor.getFilterChain().addLast(“codec”,new ProtocolCodecFilter(new AMF3CodecFactory()));
acceptor.getSessionConfig().setReadBufferSize(2048);
//设置Session空闲时间
acceptor.getSessionConfig().setIdleTime(IdleStatus.BOTH_IDLE, 60);
//设置消息处理类
acceptor.setHandler(new MessageManage());
try {
   //绑定监听端口
   acceptor.bind(new InetSocketAddress(socketPORT));
} catch (IOException e) {
   System.out.println(“服务器错误:” +e.toString()+e.getMessage());
}
System.out.println(“服务器已启动端口为:” + socketPORT);
}
}
/**
* 类功能描述:通讯服务器FLEX843端口认证解决跨域沙箱安全
* 类创建人:
* 创建时间：
*/
import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.net.ServerSocket;
import java.net.Socket;
public class AttestationServer implements Runnable {
private ServerSocket server;
private BufferedReader reader;
private BufferedWriter writer;
private static final String securityXml = “<?xml version=’1.0′?>”
                           + “<cross-domain-policy> ”
                           + “<allow-access-from domain=’*’ to-ports=’10012′/>”
                           + “</cross-domain-policy> “;
public AttestationServer() {
createServerSocket(843);
new Thread(this).start();
}
private void createServerSocket(int port) {
try {
   server = new ServerSocket(port);
} catch (IOException e) {
   System.exit(0);
}
}
public void run() {
while (true) {
   Socket client = null;
   try {
    client = server.accept();
    InputStreamReader input = new InputStreamReader(client.getInputStream(), “UTF-8″);
    reader = new BufferedReader(input);
    OutputStreamWriter output = new OutputStreamWriter(client.getOutputStream(), “UTF-8″);
    writer = new BufferedWriter(output);
    StringBuilder sb = new StringBuilder();
    int request = 0;
    while ((request = reader.read()) != -1) {
     if (request != ‘\0′) {
      sb.append((char) request);
     } else
      break;
    }
    String message = sb.toString();
    if (message.indexOf(“<policy-file-request/>”) >= 0) {
     writer.write(securityXml + “\0″);
     writer.flush();
    }
    client.close();
   } catch (Exception e) {
    e.printStackTrace();
    try {
     if (client != null) {
      client.close();
      client = null;
     }
    } catch (IOException ex) {
     ex.printStackTrace();
    } finally {
     System.gc();
    }
   }
}
}
}

package common;
/**
* 类功能描述:AMF3编解码工厂
* 类创建人:
* 创建时间：
*/
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.codec.ProtocolCodecFactory;
import org.apache.mina.filter.codec.ProtocolDecoder;
import org.apache.mina.filter.codec.ProtocolEncoder;
public class AMF3CodecFactory implements ProtocolCodecFactory {
private ProtocolEncoder encoder;
    private ProtocolDecoder decoder;
public AMF3CodecFactory() {
   encoder = new AMF3Encoder();
         decoder = new AMF3Decoder();
}
public ProtocolDecoder getDecoder(IoSession session) throws Exception {
return decoder;
}
public ProtocolEncoder getEncoder(IoSession session) throws Exception {
return encoder;
}
}
package common;
/**
* 类功能描述:对传入消息转解码
* 类创建人:
* 创建时间：
*/
import java.io.DataInputStream;
import java.util.HashMap;
import java.util.zip.InflaterInputStream;
import flex.messaging.io.SerializationContext;
import flex.messaging.io.amf.ASObject;
import flex.messaging.io.amf.Amf3Input;
import org.apache.mina.core.buffer.IoBuffer;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.codec.CumulativeProtocolDecoder;
import org.apache.mina.filter.codec.ProtocolDecoderOutput;
public class AMF3Decoder extends CumulativeProtocolDecoder {
private SerializationContext context = new SerializationContext();
private final Amf3Input amfin;
public AMF3Decoder() {
amfin = new Amf3Input(context);
}
protected boolean doDecode(IoSession session, IoBuffer in,ProtocolDecoderOutput out) throws Exception {
in.position(0);
amfin.setInputStream(new InflaterInputStream(new DataInputStream(in.asInputStream())));
Object message = amfin.readObject();
if(message instanceof ASObject){
   HashMap<?, ?> msgMap = (HashMap<?, ?>) message;
   out.write(msgMap);
   in.free();
   return true;
}
return false;
}
}
package common;
/**
* 类功能描述:对传出消息转装码
* 类创建人:
* 创建时间：
*/
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.zip.Deflater;
import entity.MessageInfo;
import flex.messaging.io.SerializationContext;
import flex.messaging.io.amf.Amf3Output;
import org.apache.mina.core.buffer.IoBuffer;
import org.apache.mina.core.session.AttributeKey;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.codec.ProtocolEncoder;
import org.apache.mina.filter.codec.ProtocolEncoderOutput;
public class AMF3Encoder implements ProtocolEncoder {
private final AttributeKey DEFLATER = new AttributeKey(getClass(), “deflater”);
private static int buffersize = 2048;
private final SerializationContext context = new SerializationContext();
private final Amf3Output amfout;

public AMF3Encoder() {
amfout = new Amf3Output(context);
}
public void encode(IoSession session, Object message,ProtocolEncoderOutput out) throws Exception {
if(message instanceof MessageInfo){
   MessageInfo messageinfo = (MessageInfo)message;//返回封装的消息对象
   IoBuffer buffer;
   ByteArrayOutputStream stream = new ByteArrayOutputStream();
   amfout.setOutputStream(stream);
   amfout.writeObject(messageinfo);
   amfout.flush();
   byte bytes[] = compress(session,stream.toByteArray());
   buffer = IoBuffer.allocate(bytes.length,false);
   buffer.put(bytes);
   buffer.flip();
   out.write(buffer);
   buffer.free();
}
}

private byte[] compress(IoSession session,byte[] inputs){
Deflater deflater = (Deflater)session.getAttribute(DEFLATER);
if(deflater == null){
   deflater = new Deflater();
   session.setAttribute(DEFLATER,deflater);
}
deflater.reset();
deflater.setInput(inputs);
deflater.finish();
byte outputs[] = new byte[0];
ByteArrayOutputStream stream = new ByteArrayOutputStream(inputs.length);
byte[] bytes = new byte[buffersize];
int value;
while(!deflater.finished()){
   value = deflater.deflate(bytes);
   stream.write(bytes,0, value);
}
outputs = stream.toByteArray();
try {
   stream.close();
} catch (IOException e) {
   e.printStackTrace();
}
return outputs;
}

public void dispose(IoSession session) throws Exception {
amfout.close();
}
}
/**
* 类功能描述:通讯消息处理
* 类创建人:
* 创建时间：
*/
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Hashtable;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import org.apache.mina.core.service.IoHandlerAdapter;
import org.apache.mina.core.session.IoSession;
import common.MessageType;
import common.UserListManage;
import entity.MessageInfo;
import entity.UserInfo;
public class MessageManage extends IoHandlerAdapter {
/**
* 方法功能描述: 消息处理
* 参数: IoSession session 接收到的消息所属IoSession
*       Object    message 接收的消息
* 返回值: void
*/
public void messageReceived(IoSession session, Object message)throws Exception {
HashMap<?,?> msgMap=(HashMap<?,?>)message;
}
}

分享到：