1// 2// ======================================================================== 3// Copyright (c) 1995-2014 Mort Bay Consulting Pty. Ltd. 4// ------------------------------------------------------------------------ 5// All rights reserved. This program and the accompanying materials 6// are made available under the terms of the Eclipse Public License v1.0 7// and Apache License v2.0 which accompanies this distribution. 8// 9// The Eclipse Public License is available at 10// http://www.eclipse.org/legal/epl-v10.html 11// 12// The Apache License v2.0 is available at 13// http://www.opensource.org/licenses/apache2.0.php 14// 15// You may elect to redistribute this code under either of these licenses. 16// ======================================================================== 17// 18 19package org.eclipse.jetty.io.nio; 20 21import java.io.IOException; 22import java.nio.ByteBuffer; 23import java.nio.channels.SelectionKey; 24import java.nio.channels.SocketChannel; 25import java.util.List; 26 27import org.eclipse.jetty.io.Buffer; 28import org.eclipse.jetty.io.NetworkTrafficListener; 29import org.eclipse.jetty.util.log.Log; 30import org.eclipse.jetty.util.log.Logger; 31 32public class NetworkTrafficSelectChannelEndPoint extends SelectChannelEndPoint 33{ 34 private static final Logger LOG = Log.getLogger(NetworkTrafficSelectChannelEndPoint.class); 35 36 private final List<NetworkTrafficListener> listeners; 37 38 public NetworkTrafficSelectChannelEndPoint(SocketChannel channel, SelectorManager.SelectSet selectSet, SelectionKey key, int maxIdleTime, List<NetworkTrafficListener> listeners) throws IOException 39 { 40 super(channel, selectSet, key, maxIdleTime); 41 this.listeners = listeners; 42 } 43 44 @Override 45 public int fill(Buffer buffer) throws IOException 46 { 47 int read = super.fill(buffer); 48 notifyIncoming(buffer, read); 49 return read; 50 } 51 52 @Override 53 public int flush(Buffer buffer) throws IOException 54 { 55 int position = buffer.getIndex(); 56 int written = super.flush(buffer); 57 notifyOutgoing(buffer, position, written); 58 return written; 59 } 60 61 @Override 62 protected int gatheringFlush(Buffer header, ByteBuffer bbuf0, Buffer buffer, ByteBuffer bbuf1) throws IOException 63 { 64 int headerPosition = header.getIndex(); 65 int headerLength = header.length(); 66 int bufferPosition = buffer.getIndex(); 67 int written = super.gatheringFlush(header, bbuf0, buffer,bbuf1); 68 notifyOutgoing(header, headerPosition, written > headerLength ? headerLength : written); 69 notifyOutgoing(buffer, bufferPosition, written > headerLength ? written - headerLength : 0); 70 return written; 71 } 72 73 public void notifyOpened() 74 { 75 if (listeners != null && !listeners.isEmpty()) 76 { 77 for (NetworkTrafficListener listener : listeners) 78 { 79 try 80 { 81 listener.opened(_socket); 82 } 83 catch (Exception x) 84 { 85 LOG.warn(x); 86 } 87 } 88 } 89 } 90 91 public void notifyIncoming(Buffer buffer, int read) 92 { 93 if (listeners != null && !listeners.isEmpty() && read > 0) 94 { 95 for (NetworkTrafficListener listener : listeners) 96 { 97 try 98 { 99 Buffer view = buffer.asReadOnlyBuffer(); 100 listener.incoming(_socket, view); 101 } 102 catch (Exception x) 103 { 104 LOG.warn(x); 105 } 106 } 107 } 108 } 109 110 public void notifyOutgoing(Buffer buffer, int position, int written) 111 { 112 if (listeners != null && !listeners.isEmpty() && written > 0) 113 { 114 for (NetworkTrafficListener listener : listeners) 115 { 116 try 117 { 118 Buffer view = buffer.asReadOnlyBuffer(); 119 view.setGetIndex(position); 120 view.setPutIndex(position + written); 121 listener.outgoing(_socket, view); 122 } 123 catch (Exception x) 124 { 125 LOG.warn(x); 126 } 127 } 128 } 129 } 130 131 public void notifyClosed() 132 { 133 if (listeners != null && !listeners.isEmpty()) 134 { 135 for (NetworkTrafficListener listener : listeners) 136 { 137 try 138 { 139 listener.closed(_socket); 140 } 141 catch (Exception x) 142 { 143 LOG.warn(x); 144 } 145 } 146 } 147 } 148} 149