Skip to content

Commit

Permalink
adopt cdo http bytes
Browse files Browse the repository at this point in the history
  • Loading branch information
KenelLiu authored and KenelLiu committed Jul 11, 2017
1 parent 5b0f322 commit 3fc9e65
Show file tree
Hide file tree
Showing 5 changed files with 109 additions and 86 deletions.
4 changes: 2 additions & 2 deletions CDOFramework/src/main/java/com/cdoframework/cdolib/servicebus/ServiceBus.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -528,7 +528,7 @@ private void onTransStarted(String strServiceName,String strTransName,CDO cdoReq
if(strTransName2.startsWith("GET") || strTransName2.startsWith("FIND") || strTransName2.startsWith("RETRIEVE")){
return;
}
logger.info(new StringBuilder().append("Starting handle ").append(" ServceName=").append(strServiceName).append(" transName=").append(strTransName).append("\r\n").append(cdoRequest.toXMLLog()).toString());
logger.info(new StringBuilder().append("Starting handle ").append(" ServceName=").append(strServiceName).append(" transName=").append(strTransName).append("\r\n").append(cdoRequest.toString()).toString());
}
}

Expand All @@ -545,7 +545,7 @@ private void onTransFinished(String strServiceName,String strTransName,CDO cdoRe
}else{
sb.append(" code=").append(retResult.getCode()).append(" text=").append(retResult.getText()).append(" info=").append(retResult.getInfo());
}
sb.append(" cdoResponse=").append(cdoResponse.toXMLLog());
sb.append(" cdoResponse=").append(cdoResponse.toString());
logger.info(sb.toString());
}
}
Expand Down
25 changes: 25 additions & 0 deletions CDOUtil/src/main/java/com/cdo/util/common/BytesStream.java
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,25 @@
package com.cdo.util.common;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import com.cdo.util.constants.Constants;

public class BytesStream {

public static byte[] readStream(InputStream in) throws IOException{
try{
ByteArrayOutputStream out=new ByteArrayOutputStream();
byte[] buf = new byte[Constants.Byte.defaultSize];
int len;
while ((len = in.read(buf)) > 0) {
out.write(buf, 0, len);
}
return out.toByteArray();
}catch(Exception ex){
throw new IOException(ex.getMessage(), ex);
}finally{
if(in!=null){try{ in.close();}catch(Exception ex){}}
}
}
}
2 changes: 1 addition & 1 deletion CDOUtil/src/main/java/com/cdo/util/constants/Constants.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,7 @@ public static class Business{
public static final String IDLE_KeepAliveTime="business.idleKeepAliveTime";
//队列长度
public static final String QueueSize="business.queueSize";//队列最最大长度是多少
//阈值 remain =taskCount-taskComplete=
//阈值 remain =taskCount-taskComplete=activeCount+queueSize(队列剩余的个数)
public static final String Threshold="business.threshold";
}

Expand Down
54 changes: 30 additions & 24 deletions CDOUtil/src/main/java/com/cdo/util/http/HttpClient.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -198,34 +198,40 @@ private void setRequestParam() {
if (this.transMode ==TRANSMODE_BODY) {
//作为body传输
if (this.body!= null) {
entity = new StringEntity(this.body,Constants.Encoding.CHARSET_UTF8);
// entity=new StringEntity(this.body, Constants.Encoding.CHARSET_UTF8);
entity = new ByteArrayEntity(this.body.getBytes(Constants.Encoding.CHARSET_UTF8));
}
} else {
//解决与OAM项目冲突问题,
if(getTransCDO()!=null){
entity = new MultipartEntity();
((MultipartEntity)entity).addPart(Constants.CDO.HTTP_CDO_REQUEST,
new StringBody(Base64.encodeBase64String(Zipper.zipBytes(Serializable.protoCDO2Byte(getTransCDO())))));
}else{
if(this.uploadFiles!=null){
//有文件传输 使用 MultipartEntity
entity=entity==null?new MultipartEntity():entity;
if(this.uploadFiles!=null){
//有文件传输 使用 MultipartEntity
entity=entity==null?new MultipartEntity():entity;
if(this.uploadFiles!=null){
for (Map.Entry<String,File> entry : this.uploadFiles.entrySet()) {
if ((entry.getKey() == null) || (entry.getKey().trim().equals("")))
continue;
((MultipartEntity)entity).addPart(entry.getKey(), new FileBody(entry.getValue()));
}
}
//设置普通一般参数
for(NameValuePair pair:this.paramsList){
((MultipartEntity)entity).addPart(pair.getName(), new StringBody(pair.getValue(),Charset.forName(Constants.Encoding.CHARSET_UTF8)));
}
for (Map.Entry<String,File> entry : this.uploadFiles.entrySet()) {
if ((entry.getKey() == null) || (entry.getKey().trim().equals("")))
continue;
((MultipartEntity)entity).addPart(entry.getKey(), new FileBody(entry.getValue()));
}
}
//设置普通一般参数
for(NameValuePair pair:this.paramsList){
((MultipartEntity)entity).addPart(pair.getName(), new StringBody(pair.getValue(),Charset.forName(Constants.Encoding.CHARSET_UTF8)));
}
if(getTransCDO()!=null){
entity = new MultipartEntity();
((MultipartEntity)entity).addPart(Constants.CDO.HTTP_CDO_REQUEST,
new StringBody(Base64.encodeBase64String(Serializable.protoCDO2Byte(getTransCDO()))));
}
}else{
//只有CDO 使用二进制传输
if(getTransCDO()!=null){
entity = new ByteArrayEntity(Serializable.protoCDO2Byte(getTransCDO()));
}else{
//普通请求
entity = new UrlEncodedFormEntity(paramsList,Charset.forName(Constants.Encoding.CHARSET_UTF8));
}
}
//普通请求
entity = new UrlEncodedFormEntity(paramsList,Charset.forName(Constants.Encoding.CHARSET_UTF8));
}
}



/**4.5
*
Expand Down
110 changes: 51 additions & 59 deletions CDOWeb/src/main/java/com/cdoframework/cdolib/web/CDOServlet.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

package com.cdoframework.cdolib.web;

import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
Expand All @@ -16,6 +17,7 @@

import javax.servlet.ServletConfig;
import javax.servlet.ServletException;
import javax.servlet.ServletInputStream;
import javax.servlet.ServletOutputStream;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
Expand All @@ -31,9 +33,7 @@
import org.apache.commons.io.FilenameUtils;
import org.apache.log4j.Logger;

import com.cdo.business.rpc.client.RPCResponse;
import com.cdo.business.threads.ThreadPoolFactory;
import com.cdo.util.common.Zipper;
import com.cdo.util.constants.Constants;
import com.cdo.util.resource.GlobalResource;
import com.cdo.util.serial.Serializable;
Expand All @@ -53,11 +53,10 @@ public abstract class CDOServlet extends HttpServlet
/**
*
*/
private static final long serialVersionUID = 6566513350461152047L;

private static final long serialVersionUID = 6566513350461152047L;
private static Logger log=Logger.getLogger(CDOServlet.class);
private final static long maxFileSize=1024*1024*50;
ThreadPoolFactory threadPool=ThreadPoolFactory.getInstatnce();

// 静态对象,所有static在此声明并初始化------------------------------------------------------------------------

// 内部对象,所有在本类中创建并使用的对象在此声明--------------------------------------------------------------
Expand Down Expand Up @@ -91,40 +90,38 @@ public void doGet(HttpServletRequest request,HttpServletResponse response) throw
public void doPost(HttpServletRequest request,HttpServletResponse response) throws ServletException,IOException
{
// 构造请求对象
CDO cdoRequest=null;
String strCDORequest=null;
CDO cdoRequest=null;
Map<String, File> mapFileMap=null;
String strTransName=null;
try
{
String serialFile=request.getHeader(Constants.CDO.HTTP_CDO_UPLOADFILE_KEY);
if(serialFile!=null && serialFile.trim().equals("1")){
//表示有文件对象需要上传
mapFileMap=new HashMap<String,File>();
strCDORequest=processUploadFile(request, mapFileMap);
}else{
cdoRequest=processProtobufEntity(request);
}

if(cdoRequest==null){
cdoRequest=CDO.fromXML(strCDORequest);
cdoRequest=processUploadFile(request, mapFileMap);
//设置文件
if(mapFileMap!=null){
for(Iterator<Map.Entry<String, File>> iterator=mapFileMap.entrySet().iterator();iterator.hasNext();){
Entry<String, File> entry=iterator.next();
cdoRequest.setFileValue(entry.getKey(), entry.getValue());
}
}
}
}else{
//http的二进制传输
cdoRequest=processProtobufEntity(request);
}

//表示异步调用 数据已经发送完毕，则表示成功,不关心回调结果,若服务器当前线程处理不过来，则异步强制使用同步
if(cdoRequest.exists(ITransService.ASYNCH_KEY) &&
cdoRequest.getBooleanValue(ITransService.ASYNCH_KEY)
&& threadPool.isAdd()){
threadPool.submit(cdoRequest);
outPutAsync(response);
return;
boolean async=cdoRequest.exists(ITransService.ASYNCH_KEY)?cdoRequest.getBooleanValue(ITransService.ASYNCH_KEY):false;
if(async){
//如果后端服务器，异步已经处理不过来前端传过来的数据,则使用同步处理，达到阻塞效果
ThreadPoolFactory threadPool=ThreadPoolFactory.getInstatnce();
if(threadPool.isAdd()){
threadPool.submit(cdoRequest);
outPutAsync(response);
return;
}
}
strTransName=cdoRequest.getStringValue("strTransName");
}catch(Exception e){
log.error("error:"+e.getMessage(),e);
outPutFail(response," Request Parameter Error :"+e.getMessage());
Expand All @@ -134,15 +131,18 @@ public void doPost(HttpServletRequest request,HttpServletResponse response) thro
CDO cdoResponse=new CDO();
Return ret=null;
try{
ret=handleTrans(request,response,cdoRequest,cdoResponse);
ret=handleTrans(request,response,cdoRequest,cdoResponse);
}catch (Throwable e){
log.error("error:"+e.getMessage(),e);
outPutFail(response," Service Internal Error :"+e.getMessage());
return;
}
if(ret==null)
{
outPutFail(response," Request method not found:strTransName="+strTransName);
String strTransName=cdoRequest.exists(ITransService.TRANSNAME_KEY)?
cdoRequest.getStringValue(ITransService.TRANSNAME_KEY):"NULL";
String strServiceName=cdoRequest.exists(ITransService.SERVICENAME_KEY)?cdoRequest.getStringValue(ITransService.SERVICENAME_KEY):"NULL";
outPutFail(response," Request method not found:strServiceName="+strServiceName+",strTransName="+strTransName);
return;
}
// 输出结果
Expand Down Expand Up @@ -290,37 +290,29 @@ public CDOServlet()
* @throws ServletException
* @throws IOException
*/
protected CDO processProtobufEntity(HttpServletRequest request) throws ServletException, IOException {
try {
DiskFileItemFactory factory = new DiskFileItemFactory();
factory.setSizeThreshold(DiskFileItemFactory.DEFAULT_SIZE_THRESHOLD);//设置成10K

ServletFileUpload upload = new ServletFileUpload(factory);
// 设置文件上传的大小限制
upload.setFileSizeMax(maxFileSize);
// 设置文件上传的头编码，如果需要正确接收中文文件路径或者文件名
// 这里需要设置对应的字符编码，为了通用这里设置为UTF-8
upload.setHeaderEncoding(Constants.Encoding.CHARSET_UTF8);
//解析请求数据包
List<FileItem> fileItems = upload.parseRequest(request);
//遍历解析完成后的Form数据和上传文件数据
for (Iterator<FileItem> iterator = fileItems.iterator(); iterator.hasNext();) {
FileItem fileItem = iterator.next();
String fieldName = fileItem.getFieldName();
if(fieldName.equals(Constants.CDO.HTTP_CDO_REQUEST)){
byte[] array=Zipper.unzipBytes(Base64.decodeBase64(fileItem.getString())) ;
return Serializable.byte2ProtoCDO(array);
}
}
throw new IOException("param["+Constants.CDO.HTTP_CDO_REQUEST+"] is not exists.... ");
} catch(Exception e) {
protected CDO processProtobufEntity(HttpServletRequest request) throws IOException {
ServletInputStream sis=null;
ByteArrayOutputStream baos=null;
try {
final int BUFFER_SIZE = 4*Constants.Byte.defaultSize;
byte[] buffer = new byte[ BUFFER_SIZE];
sis = request .getInputStream();
int length = 0;
baos = new ByteArrayOutputStream();
while(( length = sis.read( buffer))>0){
baos.write( buffer, 0, length);
}
return Serializable.byte2ProtoCDO(baos.toByteArray());
} catch(Throwable e) {
throw new IOException(e.getMessage(),e);
}

}finally{
if(sis!=null){try{sis.close();}catch(Exception ex){}}
if(baos!=null){try{baos.close();}catch(Exception ex){}}
}
}

protected String processUploadFile(HttpServletRequest request,Map<String,File> mapFile) throws ServletException, IOException {
String strCDORequest=null;
protected CDO processUploadFile(HttpServletRequest request,Map<String,File> mapFile) throws ServletException, IOException {

try {
String saveDirPath =GlobalResource.cdoConfig==null?null:GlobalResource.cdoConfig.getString(Constants.CDO.HTTP_UPLOAD_FILE_PATH);
String tmpDirPath =GlobalResource.cdoConfig==null?null:GlobalResource.cdoConfig.getString(Constants.CDO.TEMP_FILE_PATH);
Expand Down Expand Up @@ -390,16 +382,16 @@ protected String processUploadFile(HttpServletRequest request,Map<String,File> m
if(log.isDebugEnabled())
log.debug("fieldName[" + fieldName + "] value[" + fileItem.getString() + "]");
if(fieldName.equals(Constants.CDO.HTTP_CDO_REQUEST)){
strCDORequest=fileItem.getString();
byte[] array=Base64.decodeBase64(fileItem.getString()) ;
return Serializable.byte2ProtoCDO(array);
}
}
}
if(strCDORequest==null)
throw new IOException("param["+Constants.CDO.HTTP_CDO_REQUEST+"] is not exists.... ");
//未找到HTTP_CDO_REQUEST 关键参数.....
throw new IOException("param["+Constants.CDO.HTTP_CDO_REQUEST+"] is not exists.... ");
} catch(Exception e) {
log.error("解释上传文件数据失败....."+e.getMessage(), e);
throw new IOException(e.getMessage(),e);
}
return strCDORequest;
}
}
}

0 comments on commit 3fc9e65

Please sign in to comment.