Apache kafka connect - custom websocket connector

Question

The goal is :develop a custom Kafka connector that read ,messages from the websocket in a loop method. I try to give you an example on what I've realized:

I create an interface IWebsocketClientEndpoint

    public interface IWebsocketClientEndpoint {
    IWebsocketClientEndpoint Connect() ;
    void Disconnect() throws IOException;
    IWebsocketClientEndpoint addMessageHandler(IMessageHandler msgHandler);
    void SendMessage(String message) throws Exception;
    void SendMessage(ByteBuffer message) throws Exception;
    void SendMessage(Object message) throws Exception;
    boolean isOpen();
    void Dispose()throws IOException;
}

and a class that implement above interface:

@ClientEndpoint
public class WebsocketClientEndpoint implements IWebsocketClientEndpoint {

    private WebSocketContainer _container;
    private Session _userSession = null;
    private IMessageHandler _messageHandler;
    private URI _endpointURI;

    private WebsocketClientEndpoint(URI endpointURI) {
        try {
            _endpointURI = endpointURI;
            _container = ContainerProvider.getWebSocketContainer();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }


    private WebsocketClientEndpoint(URI endpointURI, int bufferSize) {
        try {
            _endpointURI = endpointURI;
            _container = ContainerProvider.getWebSocketContainer();
            _container.setDefaultMaxBinaryMessageBufferSize(bufferSize);
            _container.setDefaultMaxTextMessageBufferSize(bufferSize);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

  
    public static IWebsocketClientEndpoint Create(URI endpointURI){
        return new WebsocketClientEndpoint(endpointURI);
    }

    public static IWebsocketClientEndpoint Create(URI endpointURI,int bufferSize){
        return new WebsocketClientEndpoint(endpointURI,bufferSize);
    }

    public IWebsocketClientEndpoint Connect()  {
        try {
                      _container.connectToServer(this, _endpointURI);
            return this;
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }


    @OnOpen
    public void onOpen(Session userSession) {
        this._userSession = userSession;

       if (this._messageHandler != null) {
            this._messageHandler.handleOpen("Web socket "+ _endpointURI +" opened");}
    }


    @OnClose
    public void onClose(Session userSession, CloseReason reason) {

        this._userSession = null;

        if (this._messageHandler != null) {
            this._messageHandler.handleClose("Web socket "+ _endpointURI +" closed. Reason: " + reason.getReasonPhrase());}
    }

    public void Disconnect() throws IOException {
        CloseReason reason  = new CloseReason(CloseReason.CloseCodes.NORMAL_CLOSURE,"Web socket closed by user");

        this._userSession.close(reason);
        this._userSession = null;
        //close notification to caller
        if (this._messageHandler != null) {
            this._messageHandler.handleClose("Web socket "+ _endpointURI +" closed. Reason: " + reason.getReasonPhrase());}
    }

    @Override
    public IWebsocketClientEndpoint addMessageHandler(IMessageHandler msgHandler) {
        this._messageHandler = msgHandler;
        return  this;
    }

    @OnMessage
    public void onMessage(String message) {
        if (this._messageHandler != null) {
            this._messageHandler.handleMessage(message);
        }
    }

    @OnMessage
    public void onMessage(ByteBuffer bytes) {

        if (this._messageHandler != null) {
            this._messageHandler.handleMessage(bytes);
        }
    }

    public void SendMessage(String message) throws Exception {
        try{
           this._userSession.getAsyncRemote().sendText(message);
        }catch (Exception ex){
            throw ex;
        }
    }

    public void SendMessage(ByteBuffer message) throws Exception {
        try{
            this._userSession.getAsyncRemote().sendBinary(message);
        }catch (Exception ex){
            throw ex;
        }
    }


    public void SendMessage(Object message) throws Exception {
            this._userSession.getAsyncRemote().sendObject(message);
        }catch (Exception ex){
            throw ex;
        }
    }

    @Override
    public boolean isOpen() {
        if (this._userSession != null){
            return  this._userSession.isOpen();
        }
        return false;
    }
}

The class WebsocketClientEndpoint is dedicated to the creation of websocket and manage of connection, disconnection, send and receive message.

The goal is: how can I adapt the my websocket structure in the kafka connect structure? I could queue the message received ("public void handleMessage(String s)) from the socket in a ConcurrentLinkedQueue, and then, in the kafka connect loop method, unqueue them. But is it the best solution?

Below, the implementation of my Kafka custom connector

My kafka Connector

public class MySourceTask extends SourceTask {
IWebsocketClientEndpoint _clientEndPoint;

    @Override
    public void start(Map<String, String> props) {

                _clientEndPoint = WebsocketClientEndpoint
                        .Create(new URI(socket))
                        .Connect();

            _clientEndPoint.addMessageHandler(new IMessageHandler() {
                @Override
                public void handleMessage(String s) {

                }

                @Override
                public void handleMessage(ByteBuffer byteBuffer) {

                }

                @Override
                public void handleClose(String s) {

                }

                @Override
                public void handleOpen(String s) {

                }
            });
     
    }

    @Override
    public List<SourceRecord> poll() throws InterruptedException {
        return null;
    }

    @Override
    public void stop() {
            _clientEndPoint.Dispose();
    }
}

Thanks in advance to anyone

Answer 1

I'd suggest adding the interface to the class

extends SourceTask implements IMessageHandler

Then

_clientEndPoint.addMessageHandler(this);

And when you implement handleMessage, add those strings to some queue. Inside the poll method, you would pop data off that queue to create SourceRecord objects to return.

Inside of stop, call this.handleClose and clean up other resources.