@Override public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload) throws ChannelException { if ( access.addAndGet(1) == 1 ) txStart = System.currentTimeMillis(); long bytes = XByteBuffer.getDataPackageLength(((ChannelData)msg).getDataPackageLength()); try { super.sendMessage(destination, msg, payload); }catch ( ChannelException x ) { msgTxErr.addAndGet(1); if ( access.get() == 1 ) access.addAndGet(-1); throw x; } mbTx += (bytes*destination.length)/(1024d*1024d); mbAppTx += bytes/(1024d*1024d); if ( access.addAndGet(-1) == 0 ) { long stop = System.currentTimeMillis(); timeTx += (stop - txStart) / 1000d; if ((msgTxCnt.get() / interval) >= lastCnt) { lastCnt++; report(timeTx); } } msgTxCnt.addAndGet(1); }
@Override public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload) throws ChannelException { if (access.addAndGet(1) == 1) txStart = System.currentTimeMillis(); long bytes = XByteBuffer.getDataPackageLength(((ChannelData) msg).getDataPackageLength()); try { super.sendMessage(destination, msg, payload); } catch (ChannelException x) { msgTxErr.addAndGet(1); if (access.get() == 1) access.addAndGet(-1); throw x; } mbTx += (bytes * destination.length) / (1024d * 1024d); mbAppTx += bytes / (1024d * 1024d); if (access.addAndGet(-1) == 0) { long stop = System.currentTimeMillis(); timeTx += (stop - txStart) / 1000d; if ((msgTxCnt.get() / interval) >= lastCnt) { lastCnt++; report(timeTx); } } msgTxCnt.addAndGet(1); }
@Override public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload) throws ChannelException { try { super.sendMessage(destination, msg, payload); } catch (ChannelException cx) { FaultyMember[] mbrs = cx.getFaultyMembers(); for (int i = 0; i < mbrs.length; i++) { if (mbrs[i].getCause() != null && (!(mbrs[i].getCause() instanceof RemoteProcessException))) {// RemoteProcessException's // are // ok this.memberDisappeared(mbrs[i].getMember()); } // end if } // for throw cx; } }
@Override public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload) throws ChannelException { boolean async = (msg.getOptions() & Channel.SEND_OPTIONS_ASYNCHRONOUS) == Channel.SEND_OPTIONS_ASYNCHRONOUS; if ( async && run ) { if ( (getCurrentSize()+msg.getMessage().getLength()) > maxQueueSize ) { if ( alwaysSend ) { super.sendMessage(destination,msg,payload); return; } else { throw new ChannelException("Asynchronous queue is full, reached its limit of " + maxQueueSize +" bytes, current:" + getCurrentSize() + " bytes."); }//end if }//end if //add to queue if ( useDeepClone ) msg = (ChannelMessage)msg.deepclone(); if (!addToQueue(msg, destination, payload) ) { throw new ChannelException("Unable to add the message to the async queue, queue bug?"); } addAndGetCurrentSize(msg.getMessage().getLength()); } else { super.sendMessage(destination, msg, payload); } }
@Override public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload) throws ChannelException { //todo, optimize, if destination.length==1, then we can do //msg.setOptions(msg.getOptions() & (~getOptionFlag()) //and just send one message if (okToProcess(msg.getOptions()) ) { super.sendMessage(destination, msg, null); ChannelMessage confirmation = null; if ( deepclone ) confirmation = (ChannelMessage)msg.deepclone(); else confirmation = (ChannelMessage)msg.clone(); confirmation.getMessage().reset(); UUIDGenerator.randomUUID(false,confirmation.getUniqueId(),0); confirmation.getMessage().append(START_DATA,0,START_DATA.length); confirmation.getMessage().append(msg.getUniqueId(),0,msg.getUniqueId().length); confirmation.getMessage().append(END_DATA,0,END_DATA.length); super.sendMessage(destination,confirmation,payload); } else { //turn off two phase commit //this wont work if the interceptor has 0 as a flag //since there is no flag to turn off //msg.setOptions(msg.getOptions() & (~getOptionFlag())); super.sendMessage(destination, msg, payload); } }
@Override public synchronized void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload) throws ChannelException { if ( hold == null ) { //System.out.println("Skipping message:"+msg); hold = (ChannelMessage)msg.deepclone(); dest = new Member[destination.length]; System.arraycopy(destination,0,dest,0,dest.length); } else { //System.out.println("Sending message:"+msg); super.sendMessage(destination,msg,payload); //System.out.println("Sending message:"+hold); super.sendMessage(dest,hold,null); hold = null; dest = null; } }
@Override public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload) throws ChannelException { try { super.sendMessage(destination, msg, payload); }catch ( ChannelException cx ) { FaultyMember[] mbrs = cx.getFaultyMembers(); for ( int i=0; i<mbrs.length; i++ ) { if ( mbrs[i].getCause()!=null && (!(mbrs[i].getCause() instanceof RemoteProcessException)) ) {//RemoteProcessException's are ok this.memberDisappeared(mbrs[i].getMember()); }//end if }//for throw cx; } }
@Override public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload) throws ChannelException { if ( !okToProcess(msg.getOptions()) ) { super.sendMessage(destination, msg, payload); return; } ChannelException cx = null; for (int i=0; i<destination.length; i++ ) { try { int nr = 0; try { outLock.writeLock().lock(); nr = incCounter(destination[i]); } finally { outLock.writeLock().unlock(); } //reduce byte copy msg.getMessage().append(nr); try { getNext().sendMessage(new Member[] {destination[i]}, msg, payload); } finally { msg.getMessage().trim(4); } }catch ( ChannelException x ) { if ( cx == null ) cx = x; cx.addFaultyMember(x.getFaultyMembers()); } }//for if ( cx != null ) throw cx; }
@Override public boolean addToQueue(ChannelMessage msg, Member[] destination, InterceptorPayload payload) { final LinkObject obj = new LinkObject(msg,destination,payload); Runnable r = new Runnable() { @Override public void run() { sendAsyncData(obj); } }; executor.execute(r); return true; }
@Override public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload) throws ChannelException { int size = msg.getMessage().getLength(); boolean frag = (size>maxSize) && okToProcess(msg.getOptions()); if ( frag ) { frag(destination, msg, payload); } else { msg.getMessage().append(frag); super.sendMessage(destination, msg, payload); } }
public void frag(Member[] destination, ChannelMessage msg, InterceptorPayload payload) throws ChannelException { int size = msg.getMessage().getLength(); int count = ((size / maxSize )+(size%maxSize==0?0:1)); ChannelMessage[] messages = new ChannelMessage[count]; int remaining = size; for ( int i=0; i<count; i++ ) { ChannelMessage tmp = (ChannelMessage)msg.clone(); int offset = (i*maxSize); int length = Math.min(remaining,maxSize); tmp.getMessage().clear(); tmp.getMessage().append(msg.getMessage().getBytesDirect(),offset,length); //add the msg nr //tmp.getMessage().append(XByteBuffer.toBytes(i),0,4); tmp.getMessage().append(i); //add the total nr of messages //tmp.getMessage().append(XByteBuffer.toBytes(count),0,4); tmp.getMessage().append(count); //add true as the frag flag //byte[] flag = XByteBuffer.toBytes(true); //tmp.getMessage().append(flag,0,flag.length); tmp.getMessage().append(true); messages[i] = tmp; remaining -= length; } for ( int i=0; i<messages.length; i++ ) { super.sendMessage(destination,messages[i],payload); } }
@Override public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload) throws ChannelException { try { byte[] data = compress(msg.getMessage().getBytes()); msg.getMessage().trim(msg.getMessage().getLength()); msg.getMessage().append(data,0,data.length); getNext().sendMessage(destination, msg, payload); } catch ( IOException x ) { log.error("Unable to compress byte contents"); throw new ChannelException(x); } }
@Override public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload) throws ChannelException { if (!okToProcess(msg.getOptions())) { super.sendMessage(destination, msg, payload); return; } ChannelException cx = null; for (int i = 0; i < destination.length; i++) { try { int nr = 0; try { outLock.writeLock().lock(); nr = incCounter(destination[i]); } finally { outLock.writeLock().unlock(); } // reduce byte copy msg.getMessage().append(nr); try { getNext().sendMessage(new Member[] { destination[i] }, msg, payload); } finally { msg.getMessage().trim(4); } } catch (ChannelException x) { if (cx == null) cx = x; cx.addFaultyMember(x.getFaultyMembers()); } } // for if (cx != null) throw cx; }
@Override public boolean addToQueue(ChannelMessage msg, Member[] destination, InterceptorPayload payload) { final LinkObject obj = new LinkObject(msg, destination, payload); Runnable r = new Runnable() { @Override public void run() { sendAsyncData(obj); } }; executor.execute(r); return true; }
@Override public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload) throws ChannelException { int size = msg.getMessage().getLength(); boolean frag = (size > maxSize) && okToProcess(msg.getOptions()); if (frag) { frag(destination, msg, payload); } else { msg.getMessage().append(frag); super.sendMessage(destination, msg, payload); } }
public void frag(Member[] destination, ChannelMessage msg, InterceptorPayload payload) throws ChannelException { int size = msg.getMessage().getLength(); int count = ((size / maxSize) + (size % maxSize == 0 ? 0 : 1)); ChannelMessage[] messages = new ChannelMessage[count]; int remaining = size; for (int i = 0; i < count; i++) { ChannelMessage tmp = (ChannelMessage) msg.clone(); int offset = (i * maxSize); int length = Math.min(remaining, maxSize); tmp.getMessage().clear(); tmp.getMessage().append(msg.getMessage().getBytesDirect(), offset, length); // add the msg nr // tmp.getMessage().append(XByteBuffer.toBytes(i),0,4); tmp.getMessage().append(i); // add the total nr of messages // tmp.getMessage().append(XByteBuffer.toBytes(count),0,4); tmp.getMessage().append(count); // add true as the frag flag // byte[] flag = XByteBuffer.toBytes(true); // tmp.getMessage().append(flag,0,flag.length); tmp.getMessage().append(true); messages[i] = tmp; remaining -= length; } for (int i = 0; i < messages.length; i++) { super.sendMessage(destination, messages[i], payload); } }
@Override public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload) throws ChannelException { try { byte[] data = compress(msg.getMessage().getBytes()); msg.getMessage().trim(msg.getMessage().getLength()); msg.getMessage().append(data, 0, data.length); getNext().sendMessage(destination, msg, payload); } catch (IOException x) { log.error("Unable to compress byte contents"); throw new ChannelException(x); } }
