开通博客赚积分
发布资源赚积分
分类

源码开发语言/平台
当前位置：首页 > 源码/资料 > 中间件编程 > 查看源码
Connection.java
资源名称：blacktie-2.0.0.M3-src.zip　[点击查看]
上传用户：xfwatch
上传日期：2020-12-14
资源大小：872k
文件大小：14k
源码类别：

中间件编程

开发平台：

Java

Connection.java：源码内容
  1. /*
  2.  * JBoss, Home of Professional Open Source
  3.  * Copyright 2008, Red Hat, Inc., and others contributors as indicated
  4.  * by the @authors tag. All rights reserved.
  5.  * See the copyright.txt in the distribution for a
  6.  * full listing of individual contributors.
  7.  * This copyrighted material is made available to anyone wishing to use,
  8.  * modify, copy, or redistribute it subject to the terms and conditions
  9.  * of the GNU Lesser General Public License, v. 2.1.
  10.  * This program is distributed in the hope that it will be useful, but WITHOUT A
  11.  * WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A
  12.  * PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more details.
  13.  * You should have received a copy of the GNU Lesser General Public License,
  14.  * v.2.1 along with this distribution; if not, write to the Free Software
  15.  * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
  16.  * MA  02110-1301, USA.
  17.  */
  18. package org.jboss.blacktie.jatmibroker.xatmi;
  20. import java.util.Arrays;
  21. import java.util.HashMap;
  22. import java.util.Iterator;
  23. import java.util.Map;
  24. import java.util.Properties;
  26. import org.apache.log4j.LogManager;
  27. import org.apache.log4j.Logger;
  28. import org.jboss.blacktie.jatmibroker.core.conf.ConfigurationException;
  29. import org.jboss.blacktie.jatmibroker.core.transport.Message;
  30. import org.jboss.blacktie.jatmibroker.core.transport.Receiver;
  31. import org.jboss.blacktie.jatmibroker.core.transport.Transport;
  32. import org.jboss.blacktie.jatmibroker.core.transport.TransportFactory;
  34. /**
  35.  * This is the connection to remote Blacktie services. It must be created using
  36.  * the ConnectionFactory
  37.  * 
  38.  * @see ConnectionFactory
  39.  */
  40. public class Connection {
  42. private static final Logger log = LogManager.getLogger(Connection.class);
  44. // AVAILABLE FLAGS
  45. public static final int TPNOBLOCK = 0x00000001;
  46. public static final int TPSIGRSTRT = 0x00000002;
  47. public static final int TPNOREPLY = 0x00000004;
  48. public static final int TPNOTRAN = 0x00000008;
  49. public static final int TPTRAN = 0x00000010;
  50. public static final int TPNOTIME = 0x00000020;
  51. public static final int TPGETANY = 0x00000080;
  52. public static final int TPNOCHANGE = 0x00000100;
  53. public static final int TPCONV = 0x00000400;
  54. public static final int TPSENDONLY = 0x00000800;
  55. public static final int TPRECVONLY = 0x00001000;
  57. // ERROR CONDITIONS
  58. public static final int TPEBADDESC = 2;
  59. public static final int TPEBLOCK = 3;
  60. public static final int TPEINVAL = 4;
  61. public static final int TPELIMIT = 5;
  62. public static final int TPENOENT = 6;
  63. public static final int TPEOS = 7;
  64. public static final int TPEPROTO = 9;
  65. public static final int TPESVCERR = 10;
  66. public static final int TPESVCFAIL = 11;
  67. public static final int TPESYSTEM = 12;
  68. public static final int TPETIME = 13;
  69. public static final int TPETRAN = 14;
  70. public static final int TPGOTSIG = 15;
  71. public static final int TPEITYPE = 17;
  72. public static final int TPEOTYPE = 18;
  73. public static final int TPEEVENT = 22;
  74. public static final int TPEMATCH = 23;
  76. // SERVICE CONDITIONS
  77. public static final short TPFAIL = 0x00000001;
  78. public static final short TPSUCCESS = 0x00000002;
  80. // Events
  81. public static final long TPEV_DISCONIMM = 0x0001;
  82. public static final long TPEV_SVCERR = 0x0002;
  83. public static final long TPEV_SVCFAIL = 0x0004;
  84. public static final long TPEV_SVCSUCC = 0x0008;
  85. public static final long TPEV_SENDONLY = 0x0020;
  87. public static final int XATMI_SERVICE_NAME_LENGTH = 15;
  89. private static int nextId;
  91. private Map<String, Transport> transports = new HashMap<String, Transport>();
  93. /**
  94.  * Any local temporary queues created in this connection
  95.  */
  96. private Map<java.lang.Integer, Receiver> temporaryQueues = new HashMap<java.lang.Integer, Receiver>();
  98. private Properties properties;
  100. private Map<Integer, Session> sessions = new HashMap<Integer, Session>();
  102. private static boolean warnedTPSIGRSTRT;
  104. /**
  105.  * The connection
  106.  * 
  107.  * @param properties
  108.  * @param username
  109.  * @param password
  110.  * @throws ConnectionException
  111.  */
  112. Connection(Properties properties) {
  113. this.properties = properties;
  114. }
  116. /**
  117.  * Allocate a new buffer
  118.  * 
  119.  * @param type
  120.  *            The type of the buffer
  121.  * @param subType
  122.  *            The subtype of the buffer
  123.  * @return The new buffer
  124.  * @throws ConnectionException
  125.  */
  126. public Buffer tpalloc(String type, String subtype)
  127. throws ConnectionException {
  128. if (type == null) {
  129. throw new ConnectionException(Connection.TPEINVAL,
  130. "No type provided");
  131. } else if (type.equals("X_OCTET")) {
  132. log.debug("Initializing a new X_OCTET");
  133. return new X_OCTET();
  134. } else if (type.equals("X_C_TYPE")) {
  135. log.debug("Initializing a new X_C_TYPE");
  136. return new X_C_TYPE(subtype, properties);
  137. } else {
  138. log.debug("Initializing a new X_COMMON");
  139. return new X_COMMON(subtype, properties);
  140. }
  141. }
  143. /**
  144.  * Synchronous call
  145.  * 
  146.  * @param svc
  147.  *            The name of the service to call
  148.  * @param idata
  149.  *            The inbound data
  150.  * @param flags
  151.  *            The flags to use
  152.  * @return The returned buffer
  153.  */
  154. public Response tpcall(String svc, Buffer buffer, int len, int flags)
  155. throws ConnectionException {
  156. log.debug("tpcall");
  157. int tpacallFlags = flags;
  158. tpacallFlags &= ~TPNOCHANGE;
  159. int cd = tpacall(svc, buffer, len, tpacallFlags);
  160. return receive(cd, flags);
  161. }
  163. /**
  164.  * Asynchronous call
  165.  * 
  166.  * @param svc
  167.  *            The name of the service to call
  168.  * @param idata
  169.  *            The inbound data
  170.  * @param flags
  171.  *            The flags to use
  172.  * @return The connection descriptor
  173.  */
  174. public int tpacall(String svc, Buffer toSend, int len, int flags)
  175. throws ConnectionException {
  176. log.debug("tpacall");
  177. int toCheck = flags & ~(TPNOTRAN | TPNOREPLY | TPNOBLOCK | TPNOTIME | TPSIGRSTRT);
  178. if (toCheck != 0) {
  179. log.trace("invalid flags remain: " + toCheck);
  180. throw new ConnectionException(Connection.TPEINVAL, "Invalid flags remain: " + toCheck);
  181. }
  183. boolean hasTPSIGSTRT = (flags & TPSIGRSTRT) == TPSIGRSTRT;
  184. if (hasTPSIGSTRT && !warnedTPSIGRSTRT) {
  185. log.error("TPSIGRSTRT NOT SUPPORTED FOR SENDS OR RECEIVES");
  186. warnedTPSIGRSTRT = true;
  187. }
  188. svc = svc.substring(0, Math.min(Connection.XATMI_SERVICE_NAME_LENGTH,
  189. svc.length()));
  190. int correlationId = 0;
  191. synchronized (this) {
  192. correlationId = ++nextId;
  193. }
  194. Transport transport = getTransport(svc);
  195. Receiver endpoint = transport.createReceiver();
  196. temporaryQueues.put(correlationId, endpoint);
  197. // TODO HANDLE TRANSACTION
  198. String type = null;
  199. String subtype = null;
  200. byte[] data = null;
  201. if (toSend != null) {
  202. data = toSend.serialize();
  203. type = toSend.getType();
  204. subtype = toSend.getSubtype();
  205. if (!type.equals("X_OCTET")) {
  206. len = data.length;
  207. }
  208. }
  210. String timeToLive = properties.getProperty("TimeToLive");
  211. int ttl = 0;
  213. if (timeToLive != null) {
  214. ttl = Integer.parseInt(timeToLive) * 1000;
  215. }
  216. transport.getSender(svc).send(endpoint.getReplyTo(), (short) 0, 0,
  217. data, len, correlationId, flags, ttl, type, subtype);
  218. if ((flags & Connection.TPNOREPLY) == Connection.TPNOREPLY) {
  219. correlationId = 0;
  220. }
  221. log.debug("Returning cd: " + correlationId);
  222. return correlationId;
  223. }
  225. /**
  226.  * Cancel the outstanding asynchronous call.
  227.  * 
  228.  * @param cd
  229.  *            The connection descriptor
  230.  * @param flags
  231.  *            The flags to use
  232.  */
  233. public int tpcancel(int cd) throws ConnectionException {
  234. log.debug("tpcancel: " + cd);
  235. int toReturn = -1;
  236. Receiver endpoint = temporaryQueues.remove(cd);
  237. if (endpoint != null) {
  238. log.debug("closing endpoint");
  239. endpoint.close();
  240. log.debug("endpoint closed");
  241. toReturn = 0;
  242. } else {
  243. log.debug("No endpoint available");
  244. throw new ConnectionException(Connection.TPEBADDESC, "cd " + cd
  245. + " does not exist");
  246. }
  247. log.debug("tpcancel returning: " + toReturn);
  248. return toReturn;
  249. }
  251. /**
  252.  * Get the reply from the server
  253.  * 
  254.  * @param cd
  255.  *            The connection descriptor to use
  256.  * @param flags
  257.  *            The flags to use
  258.  * @return The response from the server
  259.  */
  260. public Response tpgetrply(int cd, int flags) throws ConnectionException {
  261. log.debug("tpgetrply: " + cd);
  262. int toCheck = flags & ~(TPGETANY | TPNOCHANGE | TPNOBLOCK | TPNOTIME | TPSIGRSTRT);
  263. if (toCheck != 0) {
  264. log.trace("invalid flags remain: " + toCheck);
  265. throw new ConnectionException(Connection.TPEINVAL, "Invalid flags remain: " + toCheck);
  266. }
  268. boolean hasTPSIGSTRT = (flags & TPSIGRSTRT) == TPSIGRSTRT;
  269. if (hasTPSIGSTRT && !warnedTPSIGRSTRT) {
  270. log.error("TPSIGRSTRT NOT SUPPORTED FOR SENDS OR RECEIVES");
  271. warnedTPSIGRSTRT = true;
  272. }
  274. Response toReturn = receive(cd, flags);
  275. Session session = sessions.remove(cd);
  276. if (session != null) {
  277. log.debug("closing session");
  278. session.close();
  279. log.debug("closed session");
  280. }
  281. log.debug("tpgetrply returning: " + toReturn);
  282. return toReturn;
  283. }
  285. /**
  286.  * Handle the initiation of a conversation with the server
  287.  * 
  288.  * @param svc
  289.  *            The name of the service
  290.  * @param idata
  291.  *            The outbound buffer
  292.  * @param flags
  293.  *            The flags to use
  294.  * @return The connection descriptor
  295.  */
  296. public Session tpconnect(String svc, Buffer toSend, int len, int flags)
  297. throws ConnectionException {
  298. log.debug("tpconnect: " + svc);
  300. svc = svc.substring(0, Math.min(Connection.XATMI_SERVICE_NAME_LENGTH,
  301. svc.length()));
  302. // Initiate the session
  303. boolean hasTPSIGSTRT = (flags & TPSIGRSTRT) == TPSIGRSTRT;
  304. if (hasTPSIGSTRT && !warnedTPSIGRSTRT) {
  305. log.error("TPSIGRSTRT NOT SUPPORTED FOR SENDS OR RECEIVES");
  306. warnedTPSIGRSTRT = true;
  307. }
  308. svc = svc.substring(0, Math.min(Connection.XATMI_SERVICE_NAME_LENGTH,
  309. svc.length()));
  310. int correlationId = 0;
  311. synchronized (this) {
  312. correlationId = nextId++;
  313. }
  314. Transport transport = getTransport(svc);
  315. Session session = new Session(properties, transport, correlationId);
  317. Receiver endpoint = session.getReceiver();
  318. // TODO HANDLE TRANSACTION
  319. String type = null;
  320. String subtype = null;
  321. byte[] data = null;
  322. if (toSend != null) {
  323. data = toSend.serialize();
  324. type = toSend.getType();
  325. subtype = toSend.getSubtype();
  326. if (!type.equals("X_OCTET")) {
  327. len = data.length;
  328. }
  329. }
  331. String timeToLive = properties.getProperty("TimeToLive");
  332. int ttl = 0;
  334. if (timeToLive != null) {
  335. ttl = Integer.parseInt(timeToLive) * 1000;
  336. }
  337. log.debug("tpconnect sending data");
  338. transport.getSender(svc).send(endpoint.getReplyTo(), (short) 0, 0,
  339. data, len, correlationId, flags | TPCONV, ttl, type, subtype);
  341. byte[] response = null;
  342. try {
  343. log.debug("tpconnect receiving data");
  344. X_OCTET odata = (X_OCTET) session.tprecv(0);
  345. response = odata.getByteArray();
  346. log.debug("tpconnect received data");
  347. } catch (ConnectionException e) {
  348. if (e.getReceived() != null) {
  349. response = ((X_OCTET) e.getReceived()).getByteArray();
  350. log.debug("Caught an exception with data", e);
  351. } else {
  352. throw new ConnectionException(e.getTperrno(),
  353. "Could not connect");
  354. }
  355. }
  356. byte[] ack = new byte[4];
  357. byte[] bytes = "ACK".getBytes();
  358. System.arraycopy(bytes, 0, ack, 0, 3);
  359. boolean connected = response == null ? false : Arrays.equals(ack,
  360. response);
  361. if (!connected) {
  362. log.error("Could not connect");
  363. session.close();
  364. throw new ConnectionException(Connection.TPETIME,
  365. "Could not connect");
  366. }
  367. session.setCreatorState(flags);
  368. temporaryQueues.put(correlationId, endpoint);
  369. sessions.put(correlationId, session);
  371. // Return a handle to allow the connection to send/receive data on
  372. return session;
  373. }
  375. /**
  376.  * Close any resources associated with this connection
  377.  * 
  378.  * @throws ConnectionException
  379.  */
  380. public void close() throws ConnectionException {
  381. log.debug("Close connection called");
  382. Iterator<Receiver> receivers = temporaryQueues.values().iterator();
  383. while (receivers.hasNext()) {
  384. Receiver receiver = receivers.next();
  385. log.debug("closing receiver");
  386. receiver.close();
  387. log.debug("closed receiver");
  388. }
  389. temporaryQueues.clear();
  390. Iterator<Transport> transports = this.transports.values().iterator();
  391. while (transports.hasNext()) {
  392. Transport transport = transports.next();
  393. log.debug("closing transport");
  394. transport.close();
  395. log.debug("closed transport");
  396. }
  397. this.transports.clear();
  398. log.debug("Close connection finished");
  399. }
  401. private Transport getTransport(String serviceName)
  402. throws ConnectionException {
  403. Transport toReturn = transports.get(serviceName);
  404. if (toReturn == null) {
  405. try {
  406. toReturn = TransportFactory.loadTransportFactory(serviceName,
  407. properties).createTransport();
  408. } catch (ConfigurationException e) {
  409. throw new ConnectionException(Connection.TPENOENT,
  410. "Could not load transport for: " + serviceName, e);
  411. }
  412. transports.put(serviceName, toReturn);
  413. }
  414. return toReturn;
  415. }
  417. private Response receive(int cd, int flags) throws ConnectionException {
  418. log.debug("receive: " + cd);
  419. Receiver endpoint = temporaryQueues.remove(cd);
  420. if (endpoint == null) {
  421. throw new ConnectionException(Connection.TPEBADDESC,
  422. "Session does not exist");
  423. }
  424. Message message = endpoint.receive(flags);
  425. // TODO WE SHOULD BE SENDING THE CONNECTION ID?
  426. Buffer buffer = null;
  427. if (message.type != null && !message.type.equals("")) {
  428. if (message.type.equals("X_OCTET")) {
  429. log.debug("Initializing a new X_OCTET");
  430. buffer = new X_OCTET(message.data);
  431. } else if (message.type.equals("X_C_TYPE")) {
  432. log.debug("Initializing a new X_C_TYPE");
  433. buffer = new X_C_TYPE(message.subtype, properties, message.data);
  434. } else {
  435. log.debug("Initializing a new X_COMMON");
  436. buffer = new X_COMMON(message.subtype, properties, message.data);
  437. }
  438. }
  439. if (message.rval == Connection.TPFAIL) {
  440. if (message.rcode == Connection.TPESVCERR) {
  441. throw new ConnectionException(Connection.TPESVCERR, 0L,
  442. message.rcode,
  443. "Got an error back from the remote service", buffer);
  444. }
  445. throw new ConnectionException(Connection.TPESVCFAIL, 0L,
  446. message.rcode, "Got a fail back from the remote service",
  447. buffer);
  448. } else {
  449. Response response = new Response(message.rval, message.rcode,
  450. buffer, message.len, message.flags);
  452. log.debug("received returned a response? "
  453. + (response == null ? "false" : "true"));
  454. return response;
  455. }
  456. }
  457. }