001/*
002 * JGrapes Event Driven Framework
003 * Copyright (C) 2017-2018 Michael N. Lipp
004 * 
005 * This program is free software; you can redistribute it and/or modify it 
006 * under the terms of the GNU Affero General Public License as published by 
007 * the Free Software Foundation; either version 3 of the License, or 
008 * (at your option) any later version.
009 * 
010 * This program is distributed in the hope that it will be useful, but 
011 * WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
012 * or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License 
013 * for more details.
014 * 
015 * You should have received a copy of the GNU Affero General Public License along 
016 * with this program; if not, see <http://www.gnu.org/licenses/>.
017 */
018
019package org.jgrapes.webconsole.base;
020
021import com.fasterxml.jackson.core.JsonFactory;
022import com.fasterxml.jackson.core.JsonParser;
023import com.fasterxml.jackson.databind.ObjectMapper;
024import com.fasterxml.jackson.databind.json.JsonMapper;
025import com.fasterxml.jackson.datatype.jdk8.Jdk8Module;
026import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
027import com.fasterxml.jackson.module.paramnames.ParameterNamesModule;
028import java.io.IOException;
029import java.lang.ref.WeakReference;
030import java.nio.CharBuffer;
031import java.util.Optional;
032import java.util.logging.Level;
033import java.util.logging.Logger;
034import org.jgrapes.core.Components;
035import org.jgrapes.core.EventPipeline;
036import org.jgrapes.io.events.Input;
037import org.jgrapes.io.util.ManagedBuffer;
038import org.jgrapes.io.util.ManagedBufferReader;
039import org.jgrapes.io.util.ThreadCleaner;
040import org.jgrapes.webconsole.base.events.JsonInput;
041
042/**
043 * Assembles {@link Input} events until a complete
044 * JSON message has been collected and then fires
045 * the message as {@link JsonInput} event.
046 */
047public class WebSocketInputSink extends Thread {
048
049    private static final Logger logger
050        = Logger.getLogger(WebSocketInputSink.class.getName());
051    @SuppressWarnings("PMD.FieldNamingConventions")
052    protected static final ObjectMapper mapper = JsonMapper.builder()
053        .addModule(new ParameterNamesModule()).addModule(new Jdk8Module())
054        .addModule(new JavaTimeModule()).build();
055
056    private final WeakReference<ConsoleConnection> channelRef;
057    private final WeakReference<EventPipeline> pipelineRef;
058    private ManagedBufferReader jsonSource;
059
060    /**
061     * Instantiates a new web socket input reader.
062     *
063     * @param wsInPipeline the ws in pipeline
064     * @param consoleChannel the web console channel
065     */
066    public WebSocketInputSink(EventPipeline wsInPipeline,
067            ConsoleConnection consoleChannel) {
068        channelRef = new WeakReference<>(consoleChannel);
069        pipelineRef = new WeakReference<>(wsInPipeline);
070        ThreadCleaner.watch(wsInPipeline, this);
071        ThreadCleaner.watch(consoleChannel, this);
072    }
073
074    /**
075     * Forward the data to the JSON decoder.
076     *
077     * @param input the data to be converted
078     * @throws IOException Signals that an I/O exception has occurred.
079     */
080    public void feed(ManagedBuffer<CharBuffer> input) throws IOException {
081        // Delayed initialization, allows adaption to buffer size.
082        if (jsonSource == null) {
083            jsonSource = new ManagedBufferReader();
084            (Components.useVirtualThreads() ? ofVirtual()
085                : ofPlatform()).start(this);
086        }
087        jsonSource.feed(input);
088    }
089
090    /**
091     * Forward the close to the JSON decoder.
092     *
093     * @throws IOException Signals that an I/O exception has occurred.
094     */
095    public void close() throws IOException {
096        if (jsonSource == null) {
097            // Never started
098            return;
099        }
100        jsonSource.close();
101        try {
102            join(1000);
103        } catch (InterruptedException e) {
104            // Just in case
105            interrupt();
106        }
107    }
108
109    @SuppressWarnings({ "PMD.AvoidLiteralsInIfCondition",
110        "PMD.AvoidInstantiatingObjectsInLoops", "PMD.OverridingThreadRun" })
111    @Override
112    public void run() {
113        @SuppressWarnings("PMD.CloseResource")
114        JsonParser parser;
115        try {
116            parser = new JsonFactory().createParser(jsonSource);
117        } catch (IOException e) {
118            logger.severe(() -> toString() + " cannot create JSON parser: "
119                + e.getMessage());
120            return;
121        }
122        while (true) {
123            JsonRpc rpc;
124            try {
125                if (parser.nextToken() == null) {
126                    break;
127                }
128                rpc = mapper.readValue(parser, WcJsonRpc.class);
129            } catch (IOException e) {
130                logger.log(Level.SEVERE, e, () -> toString()
131                    + " cannot read JSON: " + e.getMessage());
132                break;
133            }
134            // Fully decoded JSON available.
135            ConsoleConnection connection = channelRef.get();
136            EventPipeline eventPipeline = pipelineRef.get();
137            if (eventPipeline == null || connection == null) {
138                break;
139            }
140            // WebConsole connection established, check for special disconnect
141            if ("disconnect".equals(rpc.method())
142                && connection.consoleConnectionId().equals(rpc.param(0))) {
143                connection.close();
144                return;
145            }
146            // Ordinary message from web console (view) to server.
147            connection.refresh();
148            if ("keepAlive".equals(rpc.method())) {
149                continue;
150            }
151            eventPipeline.fire(new JsonInput(rpc), connection);
152        }
153    }
154
155    @Override
156    public String toString() {
157        StringBuilder res = new StringBuilder()
158            .append(Components.objectName(this)).append(" [");
159        Optional.ofNullable(channelRef.get()).ifPresentOrElse(
160            c -> res.append(c.toString()),
161            () -> res.append('?'));
162        return res.append(']').toString();
163    }
164
165}