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}