1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.eclipse.jetty.io.nio;
20
21 import java.io.IOException;
22 import java.nio.ByteBuffer;
23 import java.nio.channels.SelectionKey;
24 import java.nio.channels.SocketChannel;
25 import java.util.List;
26
27 import org.eclipse.jetty.io.Buffer;
28 import org.eclipse.jetty.io.NetworkTrafficListener;
29 import org.eclipse.jetty.util.log.Log;
30 import org.eclipse.jetty.util.log.Logger;
31
32 public 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 }