bundles/org.simantics.jdbc/src/org/simantics/jdbc/variable/JDBCSessionManager.java

   1 package org.simantics.jdbc.variable;
   2
   3 import java.io.IOException;
   4 import java.io.InputStream;
   5 import java.net.MalformedURLException;
   6 import java.net.URI;
   7 import java.net.URL;
   8 import java.nio.file.Paths;
   9 import java.sql.Connection;
  10 import java.sql.SQLException;
  11 import java.sql.Statement;
  12 import java.util.Properties;
  13 import java.util.function.Function;
  14
  15 import org.simantics.Simantics;
  16 import org.simantics.databoard.Bindings;
  17 import org.simantics.databoard.binding.error.BindingException;
  18 import org.simantics.db.ReadGraph;
  19 import org.simantics.db.common.request.ReadRequest;
  20 import org.simantics.db.exception.DatabaseException;
  21 import org.simantics.db.layer0.variable.NodeSupport;
  22 import org.simantics.jdbc.SimanticsJDBC;
  23 import org.simantics.simulator.toolkit.StandardRealm;
  24 import org.simantics.simulator.toolkit.db.StandardVariableSessionManager;
  25 import org.simantics.simulator.variable.exceptions.NodeManagerException;
  26 import org.slf4j.Logger;
  27 import org.slf4j.LoggerFactory;
  28
  29 import com.impossibl.postgres.api.jdbc.PGConnection;
  30 import com.impossibl.postgres.api.jdbc.PGNotificationListener;
  31 import com.impossibl.postgres.jdbc.PGDataSource;
  32
  33 public class JDBCSessionManager extends StandardVariableSessionManager<JDBCNode, JDBCNodeManagerSupport> {
  34
  35     private static final Logger LOGGER = LoggerFactory.getLogger(JDBCSessionManager.class);
  36     private static JDBCSessionManager INSTANCE = new JDBCSessionManager();
  37
  38     private static final String VALUE_SIMANTICS_JDBC_HOST     = "localhost";
  39     private static final int    VALUE_SIMANTICS_JDBC_PORT     = 5432;
  40     private static final String VALUE_SIMANTICS_JDBC_USER     = "simantics";
  41     private static final String VALUE_SIMANTICS_JDBC_PASSWORD = "simantics";
  42     private static final String VALUE_SIMANTICS_JDBC_DATABASE = "simantics";
  43
  44     private String channelName;
  45     private PGNotificationListener listener;
  46
  47     private PGDataSource dataSource;
  48     private Connection connection;
  49
  50     private static Properties readProperties(InputStream s) throws IOException {
  51         try (InputStream is = s) {
  52             Properties props = new Properties();
  53             props.load(is);
  54             return props;
  55         }
  56     }
  57
  58     private static Properties safeReadProperties(URL url) {
  59         try {
  60             return readProperties(url.openStream());
  61         } catch (IOException e) {
  62             LOGGER.error("Could not read props from " + url, e);
  63             return null;
  64         }
  65     }
  66
  67     private static Properties safeReadProperties(URI uri) {
  68         try {
  69             return safeReadProperties(uri.toURL());
  70         } catch (MalformedURLException e) {
  71             LOGGER.error("Could not read props from " + uri, e);
  72             return null;
  73         }
  74     }
  75
  76     private static Properties safeReadProperties(String path) {
  77         return safeReadProperties(Paths.get(path).toUri());
  78     }
  79
  80     private static Properties readProperties() {
  81         String propFile = System.getProperty(SimanticsJDBC.PROP_SIMANTICS_JDBC_PROPERTYFILE, null);
  82         if (propFile != null) {
  83             Properties p = safeReadProperties(propFile);
  84             if (p != null)
  85                 return p;
  86         }
  87         // Read default settings from built-in file and override them with values in System properties
  88         Properties p = safeReadProperties(JDBCSessionManager.class.getClassLoader().getResource("jdbc.properties"));
  89         if (p != null) {
  90             p.putAll(System.getProperties());
  91         } else {
  92             p = System.getProperties();
  93         }
  94         return p;
  95     }
  96
  97     public JDBCSessionManager() {
  98         this.channelName = "test";
  99
 100         Properties props = readProperties();
 101         String host = props.getProperty(SimanticsJDBC.PROP_SIMANTICS_JDBC_HOST, VALUE_SIMANTICS_JDBC_HOST);
 102         String port = props.getProperty(SimanticsJDBC.PROP_SIMANTICS_JDBC_PORT, "" + VALUE_SIMANTICS_JDBC_PORT);
 103         String database = props.getProperty(SimanticsJDBC.PROP_SIMANTICS_JDBC_DATABASE, VALUE_SIMANTICS_JDBC_DATABASE);
 104         String user = props.getProperty(SimanticsJDBC.PROP_SIMANTICS_JDBC_USER, VALUE_SIMANTICS_JDBC_USER);
 105         String password = props.getProperty(SimanticsJDBC.PROP_SIMANTICS_JDBC_PASSWORD, VALUE_SIMANTICS_JDBC_PASSWORD);
 106
 107         try {
 108             int portNumber = Integer.valueOf(port);
 109
 110             dataSource = new PGDataSource();
 111             dataSource.setHost(host);
 112             dataSource.setPort(portNumber);
 113             dataSource.setDatabaseName(database);
 114             dataSource.setUser(user);
 115             dataSource.setPassword(password);
 116
 117             this.connection = dataSource.getConnection();
 118             this.listener = new PGNotificationListener() {
 119                 @Override
 120                 public void notification(int processId, String channelName, String payload) {
 121                     if (LOGGER.isDebugEnabled())
 122                         LOGGER.debug("Received notification from processId={} channelName={} and payload={}", processId, channelName, payload);
 123                     Simantics.getSession().asyncRequest(new ReadRequest() {
 124                         @Override
 125                         public void run(ReadGraph graph) throws DatabaseException {
 126                             for (String realmId : INSTANCE.getRealms()) {
 127                                 try {
 128                                     JDBCRealm jdbcRealm = (JDBCRealm) INSTANCE.getOrCreateRealm(graph, realmId);
 129                                     jdbcRealm.getNodeManager().refreshVariable(new JDBCNode(payload));
 130                                 } catch (DatabaseException e) {
 131                                     LOGGER.error("Could not refresh variable in realm {} with payload {}", realmId, payload, e);
 132                                 }
 133                             }
 134                         }
 135                     });
 136                 }
 137             };
 138             createTable();
 139             init();
 140         } catch (SQLException e) {
 141             LOGGER.error("Could not initialize JDBCSessionManager!", e);
 142         }
 143     }
 144
 145     private void createTable() throws SQLException {
 146         Statement statement = connection.createStatement();
 147         statement.execute("CREATE TABLE IF NOT EXISTS simantics_table (key VARCHAR UNIQUE, value JSON)");
 148         statement.close();
 149     }
 150
 151     protected void init() throws SQLException {
 152         Statement statement = connection.createStatement();
 153         statement.execute("LISTEN " + this.channelName);
 154         statement.close();
 155         ((PGConnection) connection).addNotificationListener(this.listener);
 156     }
 157
 158     protected void destroy() throws SQLException {
 159         try (PGConnection connection = (PGConnection) dataSource.getConnection()) {
 160             Statement statement = connection.createStatement();
 161             statement.execute("UNLISTEN " + this.channelName);
 162             statement.close();
 163         }
 164     }
 165
 166
 167     @Override
 168     protected JDBCNodeManagerSupport createEngine(ReadGraph graph, String id) throws DatabaseException {
 169         return new JDBCNodeManagerSupport(id, this.dataSource, this.channelName);
 170     }
 171
 172     @Override
 173     protected StandardRealm<JDBCNode, JDBCNodeManagerSupport> createRealm(JDBCNodeManagerSupport engine, String id) {
 174         return new JDBCRealm(engine, id);
 175     }
 176
 177     public static void setValue(ReadGraph graph, String id, String key, Object value) throws DatabaseException, NodeManagerException, BindingException, InterruptedException {
 178         JDBCRealm realm = (JDBCRealm) INSTANCE.getOrCreateRealm(graph, id);
 179         realm.asyncExec(() -> {
 180             try {
 181                 realm.getNodeManager().setValue(new JDBCNode(key), key, value, Bindings.OBJECT);
 182             } catch (NodeManagerException | BindingException e) {
 183                 LOGGER.error("Could not set value {} for {}", value, key, e);
 184             }
 185         });
 186     }
 187
 188     public static NodeSupport<?> nodeSupport(ReadGraph graph, String sessionName) throws DatabaseException {
 189         return INSTANCE.getOrCreateNodeSupport(graph, sessionName);
 190     }
 191
 192     public static Object getValue(ReadGraph graph, String uri, String key) throws InterruptedException, DatabaseException {
 193         JDBCRealm realm = (JDBCRealm) INSTANCE.getOrCreateRealm(graph, uri);
 194         return realm.syncExec(new Function<Object, Object>() {
 195
 196             @Override
 197             public Object apply(Object t) {
 198                 try {
 199                     return realm.getNodeManager().getValue(new JDBCNode(key), key).getValue();
 200                 } catch (NodeManagerException e) {
 201                     LOGGER.error("Could not get value for {}", key, e);
 202                     return null;
 203                 }
 204             }
 205         });
 206     }
 207
 208 }