Start line:  
End line:  

Snippet Preview

Snippet HTML Code

Stack Overflow Questions
  /* ************************************************************************
  #
  #  DivConq
  #
  #  http://divconq.com/
  #
  #  Copyright:
  #    Copyright 2014 eTimeline, LLC. All rights reserved.
  #
 #  License:
 #    See the license.txt file in the project's top-level directory for details.
 #
 #  Authors:
 #    * Andy White
 #
 ************************************************************************ */
 package divconq.io.stream;
 
 import java.util.List;
 
 
 
 public class SplitStream extends BaseStream implements IStreamSource {
 	protected int seqnum = 1;
 	protected int size = 10 * 1024 * 1024;
 	protected String template = "file-%seq%.bin";
 	
 	protected int currchunk = 0;
     protected List<StreamMessageoutlist = new ArrayList<>();
 	
     public SplitStream() {
     }
 
 	public void init(StackEntry stackXElement el) {
 		this. = (intstack.intFromElement(el"StartAt"this.);
 		
 		String size = stack.stringFromElement(el"Size""10MB");
 		
 		this. = (int) FileUtil.parseFileSize(size);
 		
 		String temp = stack.stringFromElement(el"Template");
 		
 		if (StringUtil.isNotEmpty(temp))
 			this. = temp;
 	}
     
 	// make sure we don't return without first releasing the file reference content
     @Override
     public HandleReturn handle(TaskRun cbStreamMessage msg) {
     	if (msg == .
     		return this..handle(cbmsg);
 
     	ByteBuf in = msg.getPayload();
 
     	if (in != null) {
     		while (in.isReadable()) {
     			int amt = Math.min(in.readableBytes(), this. - this.);
     			
     			ByteBuf out = in.copy(in.readerIndex(), amt);
     			
     			in.skipBytes(amt);
     			this. += amt;
     		
     			boolean eof = (this. == this.) || (!in.isReadable() && msg.isEof());
     			
     			this..add(this.nextMessage(outmsgeof));
     			
     			if (eof) {
     				this.++;
     				this. = 0;
     			}
 			}
     		
     		in.release();
     	}
     	else if (msg.isEof()) {
 			this..add(this.nextMessage(nullmsgfalse));
     	}
     	
 		// write all messages in the queue
 		while (this..size() > 0) {
 			HandleReturn ret = this..handle(cbthis..remove(0));
 			
 			if (ret != .)
 				return ret;
 		}
     	
        	return .;
     }
     
     public StreamMessage nextMessage(ByteBuf outStreamMessage currboolean eof) {
		// create the output message
        blk.setPayload(out);
        blk.setModified(System.currentTimeMillis());		
        
        // keep the path, just vary the name to the template
        blk.setPath(curr.getPath().resolvePeer("/" + this..replace("%seq%"this. + "")));
        
        blk.setEof(eof);
        
        if (eof)
        	blk.setFileSize(this.);
        else
        	blk.setFileSize(0);						// don't know
        
        return blk;
    }
    
    @Override
    public void request(TaskRun cb) {
		// write all messages in the queue
		while (this..size() > 0) {
			HandleReturn ret = this..handle(cbthis..remove(0));
			if (ret != .)
				return;
		}
    	this..request(cb);
    }
New to GrepCode? Check out our FAQ X