Skip to content

Commit 387847c

Browse files
marci4marci4
committed
Implementation of a broadcast method
- broadcast method to send a text to a specific list of users (see TooTallNate#539) - hashCode && equals override for drafts and extensions - changed examples to new method broadcast
1 parent ceff46f commit 387847c

9 files changed

Lines changed: 139 additions & 36 deletions

File tree

‎src/main/example/ChatServer.java‎

Lines changed: 4 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -51,19 +51,19 @@ public ChatServer( InetSocketAddress address ) {
5151

5252
@Override
5353
public void onOpen( WebSocket conn, ClientHandshake handshake ) {
54-
this.sendToAll( "new connection: " + handshake.getResourceDescriptor() );
54+
broadcast( "new connection: " + handshake.getResourceDescriptor() );
5555
System.out.println( conn.getRemoteSocketAddress().getAddress().getHostAddress() + " entered the room!" );
5656
}
5757

5858
@Override
5959
public void onClose( WebSocket conn, int code, String reason, boolean remote ) {
60-
this.sendToAll( conn + " has left the room!" );
60+
broadcast( conn + " has left the room!" );
6161
System.out.println( conn + " has left the room!" );
6262
}
6363

6464
@Override
6565
public void onMessage( WebSocket conn, String message ) {
66-
this.sendToAll( message );
66+
broadcast( message );
6767
System.out.println( conn + ": " + message );
6868
}
6969

@@ -86,7 +86,7 @@ public static void main( String[] args ) throws InterruptedException , IOExcepti
8686
BufferedReader sysin = new BufferedReader( new InputStreamReader( System.in ) );
8787
while ( true ) {
8888
String in = sysin.readLine();
89-
s.sendToAll( in );
89+
s.broadcast( in );
9090
if( in.equals( "exit" ) ) {
9191
s.stop();
9292
break;
@@ -106,20 +106,4 @@ public void onStart() {
106106
System.out.println("Server started!");
107107
}
108108

109-
/**
110-
* Sends <var>text</var> to all currently connected WebSocket clients.
111-
*
112-
* @param text
113-
* The String to send across the network.
114-
* @throws InterruptedException
115-
* When socket related I/O errors occur.
116-
*/
117-
public void sendToAll( String text ) {
118-
Collection<WebSocket> con = connections();
119-
synchronized ( con ) {
120-
for( WebSocket c : con ) {
121-
c.send( text );
122-
}
123-
}
124-
}
125109
}

‎src/main/java/org/java_websocket/WebSocket.java‎

Lines changed: 10 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -28,11 +28,11 @@
2828
import java.net.InetSocketAddress;
2929
import java.nio.ByteBuffer;
3030
import java.nio.channels.NotYetConnectedException;
31+
import java.util.Collection;
3132

3233
import org.java_websocket.drafts.Draft;
3334
import org.java_websocket.framing.Framedata;
3435
import org.java_websocket.framing.Framedata.Opcode;
35-
import org.java_websocket.framing.FramedataImpl1;
3636

3737
public interface WebSocket {
3838
/**
@@ -119,13 +119,20 @@ public enum READYSTATE {
119119
* Send a frame to the other end
120120
* @param framedata the frame to send to the other end
121121
*/
122-
public abstract void sendFrame( Framedata framedata );
122+
void sendFrame( Framedata framedata );
123+
124+
/**
125+
* Send a collection of frames to the other end
126+
* @param frames the frames to send to the other end
127+
*/
128+
void sendFrame( Collection<Framedata> frames );
123129

124130
/**
125131
* Send a ping to the other end
126132
* @throws NotYetConnectedException websocket is not yet connected
127133
*/
128-
public void sendPing() throws NotYetConnectedException;
134+
void sendPing() throws NotYetConnectedException;
135+
129136
/**
130137
* Allows to send continuous/fragmented frames conveniently. <br>
131138
* For more into on this frame type see http://tools.ietf.org/html/rfc6455#section-5.4<br>

‎src/main/java/org/java_websocket/WebSocketImpl.java‎

Lines changed: 23 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -132,6 +132,11 @@ public class WebSocketImpl implements WebSocket {
132132
*/
133133
private static final Object synchronizeWriteObject = new Object();
134134

135+
/**
136+
* Attribute to cache a ping frame
137+
*/
138+
private PingFrame pingFrame;
139+
135140
/**
136141
* Creates a websocket with server role
137142
*
@@ -582,8 +587,12 @@ public void send( byte[] bytes ) throws IllegalArgumentException, WebsocketNotCo
582587
}
583588

584589
private void send( Collection<Framedata> frames ) {
585-
if( !isOpen() )
590+
if( !isOpen() ) {
586591
throw new WebsocketNotConnectedException();
592+
}
593+
if ( frames == null || frames.isEmpty() ) {
594+
throw new IllegalArgumentException();
595+
}
587596
ArrayList<ByteBuffer> outgoingFrames = new ArrayList<ByteBuffer>();
588597
for (Framedata f : frames) {
589598
if( DEBUG )
@@ -598,13 +607,21 @@ public void sendFragmentedFrame( Opcode op, ByteBuffer buffer, boolean fin ) {
598607
send( draft.continuousFrame( op, buffer, fin ) );
599608
}
600609

610+
@Override
611+
public void sendFrame(Collection<Framedata> frames) {
612+
send( frames );
613+
}
614+
601615
@Override
602616
public void sendFrame( Framedata framedata ) {
603617
send ( Collections.singletonList( framedata ) );
604618
}
605619

606620
public void sendPing() throws NotYetConnectedException {
607-
sendFrame(new PingFrame());
621+
if (pingFrame == null) {
622+
pingFrame = new PingFrame();
623+
}
624+
sendFrame(pingFrame);
608625
}
609626

610627
@Override
@@ -669,6 +686,10 @@ private void write( ByteBuffer buf ) {
669686
wsl.onWriteDemand( this );
670687
}
671688

689+
/**
690+
* Write a list of bytebuffer (frames in binary form) into the outgoing queue
691+
* @param bufs the list of bytebuffer
692+
*/
672693
private void write( List<ByteBuffer> bufs ) {
673694
synchronized ( synchronizeWriteObject ) {
674695
for (ByteBuffer b : bufs) {

‎src/main/java/org/java_websocket/client/WebSocketClient.java‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -551,6 +551,11 @@ public void sendFrame( Framedata framedata ) {
551551
engine.sendFrame( framedata );
552552
}
553553

554+
@Override
555+
public void sendFrame( Collection<Framedata> frames ) {
556+
engine.sendFrame( frames );
557+
}
558+
554559
@Override
555560
public InetSocketAddress getLocalSocketAddress() {
556561
return engine.getLocalSocketAddress();

‎src/main/java/org/java_websocket/drafts/Draft.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -167,6 +167,7 @@ protected boolean basicAccept( Handshakedata handshakedata ) {
167167
* Handle the frame specific to the draft
168168
* @param webSocketImpl the websocketimpl used for this draft
169169
* @param frame the frame which is supposed to be handled
170+
* @throws InvalidDataException will be thrown on invalid data
170171
*/
171172
public abstract void processFrame( WebSocketImpl webSocketImpl, Framedata frame ) throws InvalidDataException;
172173

‎src/main/java/org/java_websocket/drafts/Draft_10.java‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -136,12 +136,13 @@ public ByteBuffer createBinaryFrame( Framedata framedata ) {
136136
for( int i = 0; mes.hasRemaining(); i++ ) {
137137
buf.put( ( byte ) ( mes.get() ^ maskkey.get( i % 4 ) ) );
138138
}
139-
} else
139+
} else {
140140
buf.put( mes );
141-
// translateFrame ( buf.array () , buf.array ().length );
141+
//Reset the position of the bytebuffer e.g. for additional use
142+
mes.flip();
143+
}
142144
assert ( buf.remaining() == 0 ) : buf.remaining();
143145
buf.flip();
144-
145146
return buf;
146147
}
147148

‎src/main/java/org/java_websocket/drafts/Draft_6455.java‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -450,4 +450,19 @@ public String toString() {
450450
result += " extension: " + getExtension().toString();
451451
return result;
452452
}
453+
454+
@Override
455+
public boolean equals( Object o ) {
456+
if( this == o ) return true;
457+
if( o == null || getClass() != o.getClass() ) return false;
458+
459+
Draft_6455 that = ( Draft_6455 ) o;
460+
461+
return extension != null ? extension.equals( that.extension ) : that.extension == null;
462+
}
463+
464+
@Override
465+
public int hashCode() {
466+
return extension != null ? extension.hashCode() : 0;
467+
}
453468
}

‎src/main/java/org/java_websocket/extensions/DefaultExtension.java‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -87,4 +87,16 @@ public void reset() {
8787
public String toString() {
8888
return getClass().getSimpleName();
8989
}
90+
91+
@Override
92+
public int hashCode() {
93+
return getClass().hashCode();
94+
}
95+
96+
@Override
97+
public boolean equals( Object o ) {
98+
if( this == o ) return true;
99+
if( o == null ) return false;
100+
return getClass() == o.getClass();
101+
}
90102
}

‎src/main/java/org/java_websocket/server/WebSocketServer.java‎

Lines changed: 65 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -38,14 +38,7 @@
3838
import java.nio.channels.Selector;
3939
import java.nio.channels.ServerSocketChannel;
4040
import java.nio.channels.SocketChannel;
41-
import java.util.ArrayList;
42-
import java.util.Collection;
43-
import java.util.Collections;
44-
import java.util.HashSet;
45-
import java.util.Iterator;
46-
import java.util.LinkedList;
47-
import java.util.List;
48-
import java.util.Set;
41+
import java.util.*;
4942
import java.util.concurrent.BlockingQueue;
5043
import java.util.concurrent.CopyOnWriteArraySet;
5144
import java.util.concurrent.LinkedBlockingQueue;
@@ -55,6 +48,7 @@
5548
import org.java_websocket.*;
5649
import org.java_websocket.drafts.Draft;
5750
import org.java_websocket.exceptions.InvalidDataException;
51+
import org.java_websocket.exceptions.WebsocketNotConnectedException;
5852
import org.java_websocket.framing.CloseFrame;
5953
import org.java_websocket.framing.Framedata;
6054
import org.java_websocket.handshake.ClientHandshake;
@@ -777,6 +771,69 @@ public void onMessage( WebSocket conn, ByteBuffer message ) {
777771
public void onFragment( WebSocket conn, Framedata fragment ) {
778772
}
779773

774+
/**
775+
* Send a text to all connected endpoints
776+
* @param text the text to send to the endpoints
777+
*/
778+
public void broadcast(String text) {
779+
broadcast( text, connections );
780+
}
781+
782+
/**
783+
* Send a byte array to all connected endpoints
784+
* @param data the data to send to the endpoints
785+
*/
786+
public void broadcast(byte[] data) {
787+
broadcast( data, connections );
788+
}
789+
790+
/**
791+
* Send a byte array to a specific collection of websocket connections
792+
* @param data the data to send to the endpoints
793+
* @param clients a collection of endpoints to whom the text has to be send
794+
*/
795+
public void broadcast(byte[] data, Collection<WebSocket> clients) {
796+
Map<Draft, List<Framedata>> draftFrames = new HashMap<Draft, List<Framedata>>();
797+
ByteBuffer byteBufferData = ByteBuffer.wrap( data );
798+
synchronized( clients ) {
799+
for( WebSocket client : clients ) {
800+
Draft draft = client.getDraft();
801+
if( !draftFrames.containsKey( draft ) ) {
802+
List<Framedata> frames = draft.createFrames( byteBufferData, false );
803+
draftFrames.put( draft, frames );
804+
}
805+
try {
806+
client.sendFrame( draftFrames.get( draft ) );
807+
} catch ( WebsocketNotConnectedException e) {
808+
//Ignore this exception in this case
809+
}
810+
}
811+
}
812+
}
813+
814+
/**
815+
* Send a text to a specific collection of websocket connections
816+
* @param text the text to send to the endpoints
817+
* @param clients a collection of endpoints to whom the text has to be send
818+
*/
819+
public void broadcast(String text, Collection<WebSocket> clients) {
820+
Map<Draft, List<Framedata>> draftFrames = new HashMap<Draft, List<Framedata>>();
821+
synchronized( clients ) {
822+
for( WebSocket client : clients ) {
823+
Draft draft = client.getDraft();
824+
if( !draftFrames.containsKey( draft ) ) {
825+
List<Framedata> frames = draft.createFrames( text, false );
826+
draftFrames.put( draft, frames );
827+
}
828+
try {
829+
client.sendFrame( draftFrames.get( draft ) );
830+
} catch ( WebsocketNotConnectedException e) {
831+
//Ignore this exception in this case
832+
}
833+
}
834+
}
835+
}
836+
780837
/**
781838
* This class is used to process incoming data
782839
*/

0 commit comments

Comments
 (0)