Hola gente del foro, estoy tratando de hacer un módulo de notificaciones internas basadas en websocket de una aplicación, el objetivo es que un usuario al realizar un proceso administrativo la aplicación a través de websocket notifique a los usuarios registrados en el sistema y que correspondan al rol les llegue un mensaje en su bandeja de notificaciones indicando la descripción del proceso que se realizó, hasta ahora el módulo envía el mensaje a todos los usuarios conectados, pero me gustaría seleccionar solo los usuarios que indique a través de una consulta que viene de una bd. Por favor me gustaría que me ayudaran, muchas gracias. Hasta ahora este es el código que tengo:

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;

import javax.websocket.OnClose;
import javax.websocket.OnError;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;

import org.apache.catalina.websocket.MessageInbound;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.socket.server.standard.SpringConfigurator;

@ServerEndpoint(value = "/ratesrv/{clientId}", configurator = SpringConfigurator.class)
public class CustomEndPoint {

//queue holds the list of connected clients

private static Queue<Session> queue = new ConcurrentLinkedQueue<Session>();

private final static HashMap<String, CustomEndPoint> sockets = new HashMap<>();

private Session session;
private String myUniqueId;
public List<Usuario> lista = new ArrayList<Usuario>();
public String client;

public CustomEndPoint() {
super();
}

public String accion;

@Autowired
private ConsultaDao usuarioConsultaDao;

private String getMyUniqueId() {
// unique ID from this class' hash code
return Integer.toHexString(this.hashCode());
}

@OnMessage
public void onMessage(@PathParam("clientId") String clientId, Session session, String msg) {

//provided for completeness, in out scenario clients don't send any msg.

/*try {
System.out.println("received msg "+msg+" from "+clientId);
if(queue!=null){
String [] coma;
coma = msg.split(",");
String [] punto;
punto = coma[0].split(":");
accion = punto[1].replace("\"", "").replace(" ", "");
RequestGeneric request = new RequestGeneric();
request.setPropertyTobeAsked("PI_ACCION", accion);
ResponseGeneric response = usuarioConsultaDao.enviaNotificacionUsuario(request);
lista = (List<Usuario>) response.getPropertyAsked("PI_OUT_CURSOR");

sendAll(clientId, msg);
}

} catch (Exception e) {

e.printStackTrace();

}*/

}

@OnOpen
public void open(@PathParam("clientId") String clientId, Session session) {

queue.add(session);

System.out.println("New session opened: "+session.getId());
System.out.println("New client opened: "+clientId);

// save session so we can send
this.session = session;
this.setClient(clientId);

// this unique ID
this.myUniqueId = this.getMyUniqueId();

// map this unique ID to this connection
CustomEndPoint.sockets.put(this.myUniqueId, this);

// send its unique ID to the client (JSON)
this.sendClient(String.format("{\"msg\": \"uniqueId\", \"uniqueId\": \"%s\"}",
clientId));

// broadcast this new connection (with its unique ID) to all other connected clients
for (CustomEndPoint dstSocket : CustomEndPoint.sockets.values()) {
if (dstSocket == this) {
// skip me
continue;
}
dstSocket.sendClient(String.format("{\"msg\": \"newClient\", \"newClientId\": \"%s\"}",
clientId));
}

}

private void sendClient(String str) {
try {
this.session.getBasicRemote().sendText(str);
} catch (IOException e) {
e.printStackTrace();
}
}

@OnError
public void error(Session session, Throwable t) {

queue.remove(session);

System.err.println("Error on session "+session.getId());

}

@OnClose
public void closedConnection(Session session) {

queue.remove(session);

System.out.println("session closed: "+session.getId());

}

private void sendAll(@PathParam("clientId") String clientId, String msg) {

try{

List<Session> closedSession = new ArrayList();
for (Session session : queue) {

if(!session.isOpen())

{

System.err.println("Session closed: "+session.getId());

closedSession.add(session);

}

else

{
for (int i = 0; i < lista.size(); i++) {
if(lista.get(i).getUser().equalsIgnoreCase(clientId)){
session.getBasicRemote().sendText(msg);
System.out.println("Sending "+msg+" to "+i+" client(s)");
}
}

}

}

}catch (Throwable e) {

e.printStackTrace();

}

/*try {

Send the new rate to all open WebSocket sessions

ArrayList<Session > closedSessions= new ArrayList<>();

for (Session session : queue) {

if(!session.isOpen())

{

System.err.println("Closed session: "+session.getId());

closedSessions.add(session);

}

else

{

session.getBasicRemote().sendText(msg);

}

}

queue.removeAll(closedSessions);

System.out.println("Sending "+msg+" to "+queue.size()+" client(s)");
List<Usuario> listas = this.getLista();
System.out.println(listas);

} catch (Throwable e) {

e.printStackTrace();

}*/

}

public void setClient(String cliente){
client = cliente;
}

public String getClient(){
return client;
}

public List<Usuario> getLista() {
return lista;
}

public void setLista(List<Usuario> lista) {
this.lista = lista;
}

public String getAccion() {
return accion;
}

public void setAccion(String accion) {
this.accion = accion;
}

@Override
protected void onBinaryMessage(ByteBuffer arg0) throws IOException {
// TODO Auto-generated method stub

}

@Override
protected void onTextMessage(CharBuffer arg0) throws IOException {
// TODO Auto-generated method stub

}

}