001/** 002The contents of this file are subject to the Mozilla Public License Version 1.1 003(the "License"); you may not use this file except in compliance with the License. 004You may obtain a copy of the License at http://www.mozilla.org/MPL/ 005Software distributed under the License is distributed on an "AS IS" basis, 006WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License for the 007specific language governing rights and limitations under the License. 008 009The Original Code is "ConnectionHub.java". Description: 010"Provides access to shared HL7 Connections" 011 012The Initial Developer of the Original Code is University Health Network. Copyright (C) 0132001. All Rights Reserved. 014 015Contributor(s): ______________________________________. 016 017Alternatively, the contents of this file may be used under the terms of the 018GNU General Public License (the �GPL�), in which case the provisions of the GPL are 019applicable instead of those above. If you wish to allow use of your version of this 020file only under the terms of the GPL and not to allow others to use your version 021of this file under the MPL, indicate your decision by deleting the provisions above 022and replace them with the notice and other provisions required by the GPL License. 023If you do not delete the provisions above, a recipient may use your version of 024this file under either the MPL or the GPL. 025 */ 026 027package ca.uhn.hl7v2.app; 028 029import java.io.IOException; 030import java.util.Collections; 031import java.util.Map; 032import java.util.Set; 033import java.util.concurrent.ConcurrentHashMap; 034import java.util.concurrent.ConcurrentMap; 035import java.util.concurrent.ExecutorService; 036 037import ca.uhn.hl7v2.DefaultHapiContext; 038import ca.uhn.hl7v2.HL7Exception; 039import ca.uhn.hl7v2.HapiContext; 040import ca.uhn.hl7v2.HapiContextSupport; 041import ca.uhn.hl7v2.concurrent.DefaultExecutorService; 042import ca.uhn.hl7v2.llp.LowerLayerProtocol; 043import ca.uhn.hl7v2.parser.Parser; 044import ca.uhn.hl7v2.util.ReflectionUtil; 045import ca.uhn.hl7v2.util.SocketFactory; 046import org.slf4j.Logger; 047import org.slf4j.LoggerFactory; 048 049/** 050 * <p> 051 * Provides access to shared HL7 Connections. The ConnectionHub has at most one connection to any 052 * given address at any time. 053 * </p> 054 * <p> 055 * <b>Synchronization Note:</b> This class should be safe to use in a multithreaded environment. A 056 * synchronization mutex is maintained for any given target host and port, so that if two threads 057 * are trying to connect to two separate destinations neither will block, but if two threads are 058 * trying to connect to the same destination, one will block until the other has finished trying. 059 * Use caution if this class is to be used in an environment where a very large (over 1000) number 060 * of target host/port destinations will be accessed at the same time. 061 * </p> 062 * 063 * @author Bryan Tripp 064 */ 065public class ConnectionHub extends HapiContextSupport { 066 067 private static volatile ConnectionHub instance = null; 068 private static final Logger log = LoggerFactory.getLogger(ConnectionHub.class); 069 /** 070 * Set a system property with this key to a string containing an integer larger than the default 071 * ("1000") if you need to connect to a very large number of targets at the same time in a 072 * multithreaded environment. 073 */ 074 public static final String MAX_CONCURRENT_TARGETS = ConnectionHub.class.getName() + ".maxSize"; 075 private final ConcurrentMap<String, String> connectionMutexes = new ConcurrentHashMap<>(); 076 private final CountingMap<ConnectionData, Connection> connections; 077 078 /** Creates a new instance of ConnectionHub */ 079 private ConnectionHub(ExecutorService executorService) { 080 this(new DefaultHapiContext(executorService)); 081 } 082 083 private ConnectionHub(HapiContext context) { 084 super(context); 085 connections = new CountingMap<ConnectionData, Connection>() { 086 087 @Override 088 protected void dispose(Connection connection) { 089 try { 090 connection.close(); 091 } catch (IOException e) { 092 throw new RuntimeException(e); 093 } 094 } 095 096 @Override 097 protected Connection open(ConnectionData connectionData) throws Exception { 098 return ConnectionFactory 099 .open(connectionData, getHapiContext().getExecutorService()); 100 } 101 102 }; 103 } 104 105 public Set<? extends ConnectionData> allConnections() { 106 return connections.keySet(); 107 } 108 109 /** 110 * @since 2.0 111 */ 112 public Connection attach(ConnectionData data) throws HL7Exception { 113 try { 114 Connection conn; 115 // Disallow establishing same connection targets concurrently 116 connectionMutexes.putIfAbsent(data.toString(), data.toString()); 117 String mutex = connectionMutexes.get(data.toString()); 118 synchronized (mutex) { 119 discardConnectionIfStale(connections.get(data)); 120 // Create connection or increase counter 121 conn = connections.put(data); 122 } 123 return conn; 124 } catch (Exception e) { 125 log.debug("Failed to attach", e); 126 throw new HL7Exception("Cannot open connection to " + data.getHost() + ":" 127 + data.getPort() + "/" + data.getPort2(), e); 128 } 129 } 130 131 /** 132 * Returns a Connection to the given address, opening this Connection if necessary. The given 133 * Parser will only be used if a new Connection is opened, so there is no guarantee that the 134 * Connection returned will be using the Parser you provide. If you need explicit access to the 135 * Parser the Connection is using, call <code>Connection.getParser()</code>. 136 * 137 * @since 2.1 138 */ 139 public Connection attach(String host, int port, boolean tls) throws HL7Exception { 140 return attach(new ConnectionData(host, port, 0, getHapiContext().getGenericParser(), 141 getHapiContext().getLowerLayerProtocol(), tls, getHapiContext() 142 .getSocketFactory(), false)); 143 } 144 145 /** 146 * Returns a Connection to the given address, opening this Connection if necessary. The given 147 * Parser will only be used if a new Connection is opened, so there is no guarantee that the 148 * Connection returned will be using the Parser you provide. If you need explicit access to the 149 * Parser the Connection is using, call <code>Connection.getParser()</code>. 150 * 151 * @since 2.2 152 */ 153 public Connection attachLazily(String host, int port, boolean tls) throws HL7Exception { 154 return attach(new ConnectionData(host, port, 0, getHapiContext().getGenericParser(), 155 getHapiContext().getLowerLayerProtocol(), tls, getHapiContext() 156 .getSocketFactory(), true)); 157 } 158 159 /** 160 * @since 2.0 161 */ 162 public Connection attach(String host, int outboundPort, int inboundPort, boolean tls) throws HL7Exception { 163 return attach(new ConnectionData(host, outboundPort, inboundPort, getHapiContext() 164 .getGenericParser(), getHapiContext().getLowerLayerProtocol(), tls, 165 getHapiContext().getSocketFactory(), false)); 166 } 167 168 /** 169 * @since 2.2 170 */ 171 public Connection attachLazily(String host, int outboundPort, int inboundPort, boolean tls) throws HL7Exception { 172 return attach(new ConnectionData(host, outboundPort, inboundPort, getHapiContext() 173 .getGenericParser(), getHapiContext().getLowerLayerProtocol(), tls, 174 getHapiContext().getSocketFactory(), true)); 175 } 176 177 /** 178 * @since 2.0 179 */ 180 public Connection attach(String host, int outboundPort, int inboundPort, Parser parser, 181 Class<? extends LowerLayerProtocol> llpClass) throws HL7Exception { 182 return attach(host, outboundPort, inboundPort, parser, llpClass, false); 183 } 184 185 /** 186 * @since 2.0 187 */ 188 public Connection attachLazily(String host, int outboundPort, int inboundPort, Parser parser, 189 Class<? extends LowerLayerProtocol> llpClass) throws HL7Exception { 190 return attachLazily(host, outboundPort, inboundPort, parser, llpClass, false); 191 } 192 193 /** 194 * @since 2.0 195 */ 196 public Connection attach(String host, int outboundPort, int inboundPort, Parser parser, 197 Class<? extends LowerLayerProtocol> llpClass, boolean tls) throws HL7Exception { 198 LowerLayerProtocol llp = ReflectionUtil.instantiate(llpClass); 199 return attach(host, outboundPort, inboundPort, parser, llp, tls); 200 } 201 202 public Connection attachLazily(String host, int outboundPort, int inboundPort, Parser parser, 203 Class<? extends LowerLayerProtocol> llpClass, boolean tls) throws HL7Exception { 204 LowerLayerProtocol llp = ReflectionUtil.instantiate(llpClass); 205 return attachLazily(host, outboundPort, inboundPort, parser, llp, tls); 206 } 207 208 /** 209 * @since 2.0 210 */ 211 public Connection attach(String host, int outboundPort, int inboundPort, Parser parser, 212 LowerLayerProtocol llp, boolean tls) throws HL7Exception { 213 return attach(new ConnectionData(host, outboundPort, inboundPort, parser, llp, tls, null, false)); 214 } 215 216 /** 217 * @since 2.2 218 */ 219 public Connection attachLazily(String host, int outboundPort, int inboundPort, Parser parser, 220 LowerLayerProtocol llp, boolean tls) throws HL7Exception { 221 return attach(new ConnectionData(host, outboundPort, inboundPort, parser, llp, tls, null, true)); 222 } 223 224 /** 225 * @since 2.1 226 */ 227 public Connection attach(String host, int outboundPort, int inboundPort, Parser parser, LowerLayerProtocol llp, 228 boolean tls, SocketFactory socketFactory) throws HL7Exception { 229 return attach(new ConnectionData(host, outboundPort, inboundPort, parser, llp, tls, socketFactory, false)); 230 } 231 232 /** 233 * @since 2.1 234 */ 235 public Connection attachLazily(String host, int outboundPort, int inboundPort, Parser parser, LowerLayerProtocol llp, 236 boolean tls, SocketFactory socketFactory) throws HL7Exception { 237 return attach(new ConnectionData(host, outboundPort, inboundPort, parser, llp, tls, socketFactory, true)); 238 } 239 240 /** 241 * @since 2.1 242 */ 243 public Connection attach(String host, int port, Parser parser, LowerLayerProtocol llp, 244 boolean tls, SocketFactory socketFactory) throws HL7Exception { 245 return attach(new ConnectionData(host, port, 0, parser, llp, tls, socketFactory, false)); 246 } 247 248 /** 249 * @since 2.1 250 */ 251 public Connection attachLazily(String host, int port, Parser parser, LowerLayerProtocol llp, 252 boolean tls, SocketFactory socketFactory) throws HL7Exception { 253 return attach(new ConnectionData(host, port, 0, parser, llp, tls, socketFactory, true)); 254 } 255 256 /** 257 * @since 2.1 258 */ 259 public Connection attach(DefaultHapiContext hapiContext, String host, int port, boolean tls) throws HL7Exception { 260 return attach(new ConnectionData(host, port, 0, hapiContext.getGenericParser(), hapiContext.getLowerLayerProtocol(), 261 tls, hapiContext.getSocketFactory(), false)); 262 } 263 264 /** 265 * @since 2.2 266 */ 267 public Connection attachLazily(DefaultHapiContext hapiContext, String host, int port, boolean tls) throws HL7Exception { 268 return attach(new ConnectionData(host, port, 0, hapiContext.getGenericParser(), hapiContext.getLowerLayerProtocol(), 269 tls, hapiContext.getSocketFactory(), true)); 270 } 271 272 /** 273 * @since 2.1 274 */ 275 public Connection attach(DefaultHapiContext hapiContext, String host, int outboundPort, int inboundPort, boolean tls) throws HL7Exception { 276 return attach(new ConnectionData(host, outboundPort, inboundPort, hapiContext.getGenericParser(), 277 hapiContext.getLowerLayerProtocol(), tls, hapiContext.getSocketFactory(), false)); 278 } 279 280 /** 281 * @since 2.2 282 */ 283 public Connection attachLazily(DefaultHapiContext hapiContext, String host, int outboundPort, int inboundPort, boolean tls) throws HL7Exception { 284 return attach(new ConnectionData(host, outboundPort, inboundPort, hapiContext.getGenericParser(), 285 hapiContext.getLowerLayerProtocol(), tls, hapiContext.getSocketFactory(), true)); 286 } 287 288 /** 289 * @since 1.2 290 */ 291 public Connection attach(String host, int port, Parser parser, 292 Class<? extends LowerLayerProtocol> llpClass) throws HL7Exception { 293 return attach(host, port, parser, llpClass, false); 294 } 295 296 /** 297 * @since 2.0 298 */ 299 public Connection attach(String host, int port, Parser parser, 300 Class<? extends LowerLayerProtocol> llpClass, boolean tls) throws HL7Exception { 301 return attach(host, port, 0, parser, llpClass, tls); 302 } 303 304 /** 305 * @since 2.0 306 */ 307 public Connection attach(String host, int port, Parser parser, LowerLayerProtocol llp) 308 throws HL7Exception { 309 return attach(host, port, 0, parser, llp, false); 310 } 311 312 /** 313 * @since 2.2 314 */ 315 public Connection attachLazily(String host, int port, Parser parser, LowerLayerProtocol llp) 316 throws HL7Exception { 317 return attachLazily(host, port, 0, parser, llp, false); 318 } 319 320 321 /** 322 * @since 2.0 323 */ 324 public Connection attach(String host, int port, Parser parser, LowerLayerProtocol llp, 325 boolean tls) throws HL7Exception { 326 return attach(host, port, 0, parser, llp, tls); 327 } 328 329 /** 330 * @since 2.0 331 */ 332 public Connection attachLazily(String host, int port, Parser parser, LowerLayerProtocol llp, 333 boolean tls) throws HL7Exception { 334 return attachLazily(host, port, 0, parser, llp, tls); 335 } 336 337 /** 338 * Informs the ConnectionHub that you are done with the given Connection - if no other code is 339 * using it, it will be closed, so you should not attempt to use a Connection after detaching 340 * from it. If the connection is not enlisted, this method does nothing. 341 */ 342 public void detach(Connection c) { 343 ConnectionData cd = connections.find(c); 344 if (cd != null) 345 connections.remove(cd); 346 } 347 348 /** 349 * Closes and discards the given Connection so that it can not be returned in subsequent calls 350 * to attach(). This method is to be used when there is a problem with a Connection, e.g. socket 351 * connection closed by remote host. 352 */ 353 public void discard(Connection c) { 354 ConnectionData cd = connections.find(c); 355 if (cd != null) 356 connections.removeAllOf(cd); 357 } 358 359 /** 360 * Closes and discards all connections. 361 */ 362 public void discardAll() { 363 for (ConnectionData cd : allConnections()) { 364 connections.removeAllOf(cd); 365 } 366 } 367 368 private void discardConnectionIfStale(Connection conn) { 369 if (conn != null && !conn.isOpen()) { 370 log.info("Discarding connection which appears to be closed. Remote addr: {}", 371 conn.getRemoteAddress()); 372 discard(conn); 373 } 374 } 375 376 public Connection getKnownConnection(ConnectionData key) { 377 return connections.get(key); 378 } 379 380 public boolean isOpen(ConnectionData key) { 381 return getKnownConnection(key).isOpen(); 382 } 383 384 /** 385 * Returns the singleton instance of ConnectionHub 386 * 387 * @deprecated Use {@link HapiContext#getConnectionHub()} to get an instance of ConnectionHub. 388 * See <a href="http://hl7api.sourceforge.net/xref/ca/uhn/hl7v2/examples/SendAndReceiveAMessage.html">this example page</a> for an example of how to use ConnectionHub. 389 */ 390 public static ConnectionHub getInstance() { 391 return getInstance(DefaultExecutorService.getDefaultService()); 392 } 393 394 /** 395 * Returns the singleton instance of ConnectionHub. 396 * 397 * @deprecated Use {@link HapiContext#getConnectionHub()} to get an instance of ConnectionHub. 398 * See <a href="http://hl7api.sourceforge.net/xref/ca/uhn/hl7v2/examples/SendAndReceiveAMessage.html">this example page</a> for an example of how to use ConnectionHub. 399 */ 400 public synchronized static ConnectionHub getInstance(ExecutorService service) { 401 if (instance == null || service.isShutdown()) { 402 instance = new ConnectionHub(service); 403 } 404 return instance; 405 } 406 407 /** 408 * Returns the singleton instance of ConnectionHub. 409 * 410 * @deprecated Use {@link HapiContext#getConnectionHub()} to get an instance of ConnectionHub. 411 * See <a href="http://hl7api.sourceforge.net/xref/ca/uhn/hl7v2/examples/SendAndReceiveAMessage.html">this example page</a> for an example of how to use ConnectionHub. 412 */ 413 public static ConnectionHub getInstance(HapiContext context) { 414 if (instance == null || context.getExecutorService().isShutdown()) { 415 instance = new ConnectionHub(context); 416 } 417 return instance; 418 } 419 420 /** 421 * <p> 422 * Returns a new (non-singleton) instance of the ConnectionHub which uses the given executor 423 * service. 424 * </p> 425 * <p> 426 * See <a href="http://hl7api.sourceforge.net/xref/ca/uhn/hl7v2/examples/SendAndReceiveAMessage.html">this example page</a> 427 * for an example of how to use ConnectionHub. 428 * </p> 429 */ 430 public synchronized static ConnectionHub getNewInstance(HapiContext context) { 431 return new ConnectionHub(context); 432 } 433 434 /** 435 * @deprecated default executor service is shut down automatically 436 */ 437 public static void shutdown() { 438 ConnectionHub hub = getInstance(); 439 if (DefaultExecutorService.isDefaultService(hub.getHapiContext().getExecutorService())) { 440 hub.getHapiContext().getExecutorService().shutdown(); 441 instance = null; 442 } 443 } 444 445 /** 446 * Helper class that implements a map that increases/decreases a counter when an entry is 447 * added/removed. It is furthermore intended that an entry's value is derived from its key. 448 * 449 * @param <K> key class 450 * @param <D> managed value class 451 */ 452 private abstract static class CountingMap<K, D> { 453 private final Map<K, Count> content; 454 455 public CountingMap() { 456 super(); 457 content = new ConcurrentHashMap<>(); 458 } 459 460 protected abstract void dispose(D value); 461 462 public K find(D value) { 463 for (Map.Entry<K, Count> entry : content.entrySet()) { 464 if (entry.getValue().getValue().equals(value)) { 465 return entry.getKey(); 466 } 467 } 468 return null; 469 } 470 471 public D get(K key) { 472 return content.containsKey(key) ? content.get(key).getValue() : null; 473 } 474 475 public Set<K> keySet() { 476 return Collections.unmodifiableSet(content.keySet()); 477 } 478 479 protected abstract D open(K key) throws Exception; 480 481 /** 482 * If the key exists, the counter is increased. Otherwise, a value is created, and the 483 * key/value pair is added to the map. 484 */ 485 public D put(K key) throws Exception { 486 if (content.containsKey(key)) { 487 return content.put(key, content.get(key).increase()).getValue(); 488 } else { 489 Count c = new Count(open(key)); 490 content.put(key, c); 491 return c.getValue(); 492 } 493 } 494 495 /** 496 * If the counter of the key/value is greater than one, the counter is decreased. Otherwise, 497 * the entry is removed and the value is cleaned up. 498 */ 499 public D remove(K key) { 500 Count pair = content.get(key); 501 if (pair == null) 502 return null; 503 if (pair.isLast()) { 504 return removeAllOf(key); 505 } 506 return content.put(key, content.get(key).decrease()).getValue(); 507 } 508 509 /** 510 * The key/value entry is removed and the value is cleaned up. 511 */ 512 public D removeAllOf(K key) { 513 D removed = content.remove(key).value; 514 dispose(removed); 515 return removed; 516 } 517 518 private class Count { 519 private final int count; 520 private final D value; 521 522 public Count(D value) { 523 this(value, 1); 524 } 525 526 private Count(D value, int number) { 527 this.value = value; 528 this.count = number; 529 } 530 531 Count decrease() { 532 return !isLast() ? new Count(value, count - 1) : null; 533 } 534 535 public D getValue() { 536 return value; 537 } 538 539 Count increase() { 540 return new Count(value, count + 1); 541 } 542 543 boolean isLast() { 544 return count == 1; 545 } 546 547 } 548 549 } 550 551}