BlockingQueue loses its reference and throws NullPointerException onMessage Jetty WebSocket

Question

Had to ask this question because it's been a day trying to solve the problem and cannot.

I'm working with Netbeans 8.2 and java 8.

Topology:

WebSocket client on the browser
Jetty WebSocket server (java app with swing GUI)

Objective: to send data from client to server and show the data on JTextArea (GUI)

Main.java (GUI)

public class Main extends javax.swing.JPanel {

private WebSocketSwing websocketserver;
private BlockingQueue<String> stack = new ArrayBlockingQueue<String>(3);

public Main() {
    initComponents();
    // WebSocketServer
    websocketserver = new WebSocketSwing(stack);
    websocketserver.start();
    
    consumer.start();
}

Thread consumer = new Thread(new Runnable() {
    @Override
    public void run() {
        try{
            String msg;
            //consuming messages until exit message is received
            while((msg = stack.take()) !="exit"){
                Thread.sleep(10);
                System.out.println("Consumed: " + msg);
            }
        }catch(InterruptedException e) {
            e.printStackTrace();
        }
    }
});

private void initComponents() {
    //GUI code goes here
}

    public static void main(String[] args) {
    JFrame frame = new JFrame("Main GUI");
            
    java.awt.EventQueue.invokeLater(new Runnable() {
        public void run() {
            frame.setDefaultCloseOperation(javax.swing.WindowConstants.EXIT_ON_CLOSE);
            frame.getContentPane().add(new SDG());
            frame.pack();
            frame.setVisible(true);        
        }
    });   
}

private javax.swing.JLabel jLabel1;
private javax.swing.JScrollPane jScrollPane1;
private javax.swing.JTextArea txt_area;

}

WebSocketSwing class

public class WebSocketSwing extends Thread {

/**
 * @param args the command line arguments
 */
private BlockingQueue<String> stack;

public WebSocketSwing(BlockingQueue<String> queue){
    this.stack = queue;
}
@Override
public void run(){
    super.run();
    
    try {
        Server server = new Server(2014);
        WSHandler mHandler = new WSHandler();
        mHandler.SetStack(stack);
        server.setHandler(mHandler);
        server.setStopTimeout(0);
        server.start();
        //
        server.join();
        } catch (Exception e) {
            e.printStackTrace();
        }
}
}

WSHandler class

@WebSocket
public class WSHandler extends WebSocketHandler {
    private Session session;
    public BlockingQueue<String> stack = new ArrayBlockingQueue<String>(3); // **THIS instantiation should not be needed...** 
    private static ArrayList<WSHandler> sessions = new ArrayList<WSHandler>();
    
    public static ArrayList<WSHandler> getAllSessions() {
        return sessions;
    }

    /*WSHandler(BlockingQueue<String> stack) { // **I tried to send/assign the queue from the constructor but the method is not overridable**
        SetStack(stack);  // or this.stack = stack;
    }*/
    
    
    public void SetStack(BlockingQueue<String> queue){
        this.stack = queue;
        //Testing operations to see the reference to the queue was successfully passed
        System.out.println(stack.remainingCapacity());
        stack.offer("Something");  //**consumes just fine in the other Thread...**
        
    }
    
    @OnWebSocketClose
    public void onClose(int StatusCode, String reason){
        sessions.remove(this);
        System.out.println("Close: Status Code: " + StatusCode + ", reason: " + reason + ", sessions = " + sessions.size());
    }
    
    @OnWebSocketError
    public void onError(Throwable t) {
        System.out.println("Error: " + t.getMessage());
    }

    @OnWebSocketConnect
    public void onConnect(Session localSession) {
        session = localSession;
        sessions.add(this);
        
        System.out.println("Connect: " + session.getRemoteAddress().getAddress());
    }
    
    @OnWebSocketMessage
    public void onMessage(String message) {
        try {
            System.out.println("Message: " + message);
            session.getRemote().sendString("ACK");
            
            SetData(message);
            if(message.equals("exit")){
                System.out.println("Message: Bye!...");
                System.exit(0);
            }
        } catch (IOException ex) {
            Logger.getLogger(WSHandler.class.getName()).log(Level.SEVERE, null, ex);
        } 
    }
    
    private void SetData(String message){
        try{
        if (stack.offer(message)){
            System.out.print("Inserted");
        } else {
            System.out.print("NOT Inserted");
        }
        } catch(NullPointerException e){
            e.printStackTrace();
        }
    }

    @Override
    public void configure(WebSocketServletFactory factory) {
        factory.register(WSHandler.class);
    }
    
}

Result> seems like this.stack lost its reference to the queue... as if I do not initialize the BlockingQueue inside the class throws the NPE outside the SetStack Method...

Trace (when I do not initialize the BlockingQueue on WSHandler class) Situation if my understanding is correct if the reference from the Main class has been passed correctly I should not need to initialize the BlockingQueue in the Handler... Then I think that's the issue to be solved...

The NullPointerException is thrown because the object lost its reference (the one it had in the SetStack method...)... that reason is what I haven't been able to find...

2021-06-26 15:35:41.990:INFO::Thread-2: Logging initialized @470ms to org.eclipse.jetty.util.log.StdErrLog
3
2021-06-26 15:35:42.077:INFO:oejs.Server:Thread-2: jetty-9.4.42.v20210604; built: 2021-06-04T17:33:38.939Z; git: 5cd5e6d2375eeab146813b0de9f19eda6ab6e6cb; jvm 1.8.0_111-b14
Consumed: Something
2021-06-26 15:35:42.827:INFO:oejs.AbstractConnector:Thread-2: Started ServerConnector@98f7e6f{HTTP/1.1, (http/1.1)}{0.0.0.0:2014}
2021-06-26 15:35:42.830:INFO:oejs.Server:Thread-2: Started @1314ms
Connect: /127.0.0.1
Message: sample message
Happened
java.lang.NullPointerException
    at websocketswing.WSHandler.SetData(WSHandler.java:100)
    at websocketswing.WSHandler.onMessage(WSHandler.java:78)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at org.eclipse.jetty.websocket.common.events.annotated.CallableMethod.call(CallableMethod.java:70)
    at org.eclipse.jetty.websocket.common.events.annotated.OptionalSessionCallableMethod.call(OptionalSessionCallableMethod.java:72)
    at org.eclipse.jetty.websocket.common.events.JettyAnnotatedEventDriver.onTextMessage(JettyAnnotatedEventDriver.java:301)
    at org.eclipse.jetty.websocket.common.message.SimpleTextMessage.messageComplete(SimpleTextMessage.java:69)
    at org.eclipse.jetty.websocket.common.events.AbstractEventDriver.appendMessage(AbstractEventDriver.java:67)
    at org.eclipse.jetty.websocket.common.events.JettyAnnotatedEventDriver.onTextFrame(JettyAnnotatedEventDriver.java:287)
    at org.eclipse.jetty.websocket.common.events.AbstractEventDriver.incomingFrame(AbstractEventDriver.java:152)
    at org.eclipse.jetty.websocket.common.WebSocketSession.incomingFrame(WebSocketSession.java:326)
    at org.eclipse.jetty.websocket.common.extensions.AbstractExtension.nextIncomingFrame(AbstractExtension.java:148)
    at org.eclipse.jetty.websocket.common.extensions.compress.PerMessageDeflateExtension.nextIncomingFrame(PerMessageDeflateExtension.java:111)
    at org.eclipse.jetty.websocket.common.extensions.compress.CompressExtension.forwardIncoming(CompressExtension.java:169)
    at org.eclipse.jetty.websocket.common.extensions.compress.PerMessageDeflateExtension.incomingFrame(PerMessageDeflateExtension.java:90)
    at org.eclipse.jetty.websocket.common.extensions.ExtensionStack.incomingFrame(ExtensionStack.java:202)
    at org.eclipse.jetty.websocket.common.Parser.notifyFrame(Parser.java:225)
    at org.eclipse.jetty.websocket.common.Parser.parseSingleFrame(Parser.java:259)
    at org.eclipse.jetty.websocket.common.io.AbstractWebSocketConnection.onFillable(AbstractWebSocketConnection.java:459)
    at org.eclipse.jetty.websocket.common.io.AbstractWebSocketConnection.onFillable(AbstractWebSocketConnection.java:440)
    at org.eclipse.jetty.io.AbstractConnection$ReadCallback.succeeded(AbstractConnection.java:311)
    at org.eclipse.jetty.io.FillInterest.fillable(FillInterest.java:105)
    at org.eclipse.jetty.io.ChannelEndPoint$1.run(ChannelEndPoint.java:104)
    at org.eclipse.jetty.util.thread.strategy.EatWhatYouKill.runTask(EatWhatYouKill.java:338)
    at org.eclipse.jetty.util.thread.strategy.EatWhatYouKill.doProduce(EatWhatYouKill.java:315)
    at org.eclipse.jetty.util.thread.strategy.EatWhatYouKill.tryProduce(EatWhatYouKill.java:173)
    at org.eclipse.jetty.util.thread.strategy.EatWhatYouKill.produce(EatWhatYouKill.java:137)
    at org.eclipse.jetty.util.thread.QueuedThreadPool.runJob(QueuedThreadPool.java:882)
    at org.eclipse.jetty.util.thread.QueuedThreadPool$Runner.run(QueuedThreadPool.java:1036)
    at java.lang.Thread.run(Thread.java:745)

Trace (when initializing the Queue)

2021-06-26 15:39:36.821:INFO::Thread-2: Logging initialized @470ms to org.eclipse.jetty.util.log.StdErrLog
3
2021-06-26 15:39:36.889:INFO:oejs.Server:Thread-2: jetty-9.4.42.v20210604; built: 2021-06-04T17:33:38.939Z; git: 5cd5e6d2375eeab146813b0de9f19eda6ab6e6cb; jvm 1.8.0_111-b14
Consumed: Something
2021-06-26 15:39:37.961:INFO:oejs.AbstractConnector:Thread-2: Started ServerConnector@358d4f07{HTTP/1.1, (http/1.1)}{0.0.0.0:2014}
2021-06-26 15:39:37.964:INFO:oejs.Server:Thread-2: Started @1615ms
Connect: /127.0.0.1
Message: sample message
Happened
Inserted
Message: sample message
Happened
NOT Inserted
Message: sample message
Happened
NOT Inserted

Hence, I've assumed the queue loses its reference because "this queue" never gets consumed by the Consumer thread (as it does in the first assignment)

Hope someone can see what I'm not seeing...

Best regards,

updated the question, would appreciate if you re-open it... thanks — Andrew, Jun 26 '21 at 22:00
@Andrew a new WSHandler gets created with each websocket connection (because of how your `configure()` method registers it). Restructure your code, use a custom `WebSocketCreator` instead in your `configure()`, and a new class for each new websocket connection. See option 2 in prior answer for example - https://stackoverflow.com/questions/15646213/how-do-i-access-instantiated-websockets-in-jetty-9/15649791#15649791 — Joakim Erdfelt, Jun 27 '21 at 01:41
@JoakimErdfelt thanks a lot! for real!... that was what I was missing... that's the proper way to create a WebSocket with a custom constructor... man, really God only knows how much you helped me. Thanks again :) — Andrew, Jun 30 '21 at 08:02

score 2 · Accepted Answer · answered Jun 30 '21 at 12:33

A new WSHandler instance gets created with each new (and accepted/upgraded) WebSocket connection because of how you registered it in your configure() method ...

@Override
public void configure(WebSocketServletFactory factory) {
    factory.register(WSHandler.class);
}

Restructure your code.

Start by separating the WSHandler from the WebSocket Endpoint.

Make the new MyEndpoint have a constructor (or setter) for your queue object.

@WebSocket
public class MyEndpoint {
   private .... queue;

   public MyEndpoint(... queue) {
       this.queue = queue;
   }

   @OnWebSocketMessage
   public void onMessage(String str) {
       this.queue.offer(str);
   }
}

Next, you want to create a custom org.eclipse.jetty.websocket.servlet.WebSocketCreator of your own design that creates the WebSocket endpoint instance, populates it, and then hands it back to the Jetty implementation.

public static class MyWebSocketCreator implements WebSocketCreator {
    private ... masterQueue = new ...;

    @Override
    public Object createWebSocket(ServletUpgradeRequest req, ServletUpgradeResponse resp) {
        return new MyEndpoint(masterQueue);
    }
}

Lastly, you want to make your configure() method use this new creator.

@Override
public void configure(WebSocketServletFactory factory) {
    factory.setCreator(new MyWebSocketCreator());
}

This is covered in my prior answer, option 2, at How do I access instantiated WebSockets in Jetty 9?

BlockingQueue loses its reference and throws NullPointerException onMessage Jetty WebSocket

1 Answers1