001: /**
002: * Licensed to the Apache Software Foundation (ASF) under one or more
003: * contributor license agreements. See the NOTICE file distributed with
004: * this work for additional information regarding copyright ownership.
005: * The ASF licenses this file to You under the Apache License, Version 2.0
006: * (the "License"); you may not use this file except in compliance with
007: * the License. You may obtain a copy of the License at
008: *
009: * http://www.apache.org/licenses/LICENSE-2.0
010: *
011: * Unless required by applicable law or agreed to in writing, software
012: * distributed under the License is distributed on an "AS IS" BASIS,
013: * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
014: * See the License for the specific language governing permissions and
015: * limitations under the License.
016: */package org.apache.openejb.core.timer;
017:
018: import javax.sql.DataSource;
019: import java.sql.Connection;
020: import java.sql.PreparedStatement;
021: import java.sql.ResultSet;
022: import java.sql.SQLException;
023: import java.util.ArrayList;
024: import java.util.Collection;
025: import java.util.Date;
026: import java.io.ByteArrayOutputStream;
027: import java.io.ObjectOutputStream;
028: import java.io.IOException;
029: import java.io.ByteArrayInputStream;
030: import java.io.ObjectInputStream;
031:
032: import org.apache.openejb.util.LogCategory;
033: import org.apache.openejb.util.Logger;
034: import org.apache.openejb.util.Base64;
035:
036: public class DatabaseTimerStore implements TimerStore {
037: private static final Logger log = Logger.getInstance(
038: LogCategory.TIMER, "org.apache.openejb.util.resources");
039:
040: private static final String createSequenceSQL = "create sequence timertasks_seq";
041: private static final String createTableSQLWithSequence = "create table timertasks (id long primary key, serverid varchar(256) not null, timerkey varchar(256) not null, userid varchar(4096), userinfo varchar(4096), firsttime long not null, period long)";
042: private static final String createTableSQLWithIdentity = "create table timertasks (id INTEGER NOT NULL GENERATED ALWAYS AS IDENTITY (START WITH 1, INCREMENT BY 1), serverid varchar(256) not null, timerkey varchar(256) not null, userid varchar(4096), userinfo varchar(4096), firsttime NUMERIC(18,0) not null, period NUMERIC(18, 0))";
043: private static final String sequenceSQL = "select timertasks_seq.nextval";
044: private static final String identitySQL = "values IDENTITY_VAL_LOCAL()";
045: private static final String insertSQLWithSequence = "insert into timertasks (id, serverid, timerkey, userid, userinfo, firsttime, period) values (?, ?, ?, ?, ?, ?, ?)";
046: private static final String insertSQLWithIdentity = "insert into timertasks (serverid, timerkey, userid, userinfo, firsttime, period) values (?, ?, ?, ?, ?, ?)";
047: private static final String deleteSQL = "delete from timertasks where id=?";
048: private static final String selectSQL = "select id, userid, userinfo, firsttime, period from timertasks where serverid = ? and timerkey=?";
049: private static final String fixedRateUpdateSQL = "update timertasks set firsttime = ? where id = ?";
050:
051: private final String serverUniqueId;
052: private final DataSource dataSource;
053: private boolean useSequence = false;
054:
055: protected DatabaseTimerStore(String serverUniqueId,
056: DataSource datasource, boolean useSequence)
057: throws SQLException {
058: this .serverUniqueId = serverUniqueId;
059: this .dataSource = datasource;
060: this .useSequence = useSequence;
061: if (this .useSequence) {
062: execSQL(createSequenceSQL);
063: execSQL(createTableSQLWithSequence);
064: } else {
065: execSQL(createTableSQLWithIdentity);
066: }
067: }
068:
069: public TimerData getTimer(String deploymentId, long timerId) {
070: // todo not implemented
071: return null;
072: }
073:
074: public Collection<TimerData> getTimers(String deploymentId) {
075: // todo not implemented
076: return null;
077: }
078:
079: public TimerData createTimer(EjbTimerServiceImpl timerService,
080: String deploymentId, Object primaryKey, Object info,
081: Date expiration, long intervalDuration)
082: throws TimerStoreException {
083: boolean threwException = false;
084: Connection c = getConnection();
085: long id;
086: try {
087: if (useSequence) {
088: PreparedStatement seqStatement = c
089: .prepareStatement(sequenceSQL);
090: try {
091: ResultSet seqRS = seqStatement.executeQuery();
092: try {
093: seqRS.next();
094: id = seqRS.getLong(1);
095: } finally {
096: seqRS.close();
097: }
098: } finally {
099: seqStatement.close();
100: }
101: PreparedStatement insertStatement = c
102: .prepareStatement(insertSQLWithSequence);
103: try {
104: String serializedPrimaryKey = serializeObject(primaryKey);
105: String serializedInfo = serializeObject(info);
106: insertStatement.setLong(1, id);
107: insertStatement.setString(2, serverUniqueId);
108: insertStatement.setString(3, deploymentId);
109: insertStatement.setString(4, serializedPrimaryKey);
110: insertStatement.setString(5, serializedInfo);
111: insertStatement.setLong(6, expiration.getTime());
112: insertStatement.setLong(7, intervalDuration);
113: int result = insertStatement.executeUpdate();
114: if (result != 1) {
115: throw new TimerStoreException(
116: "Could not insert!");
117: }
118: } finally {
119: insertStatement.close();
120: }
121: } else {
122: PreparedStatement insertStatement = c
123: .prepareStatement(insertSQLWithIdentity);
124: try {
125: String serializedPrimaryKey = serializeObject(primaryKey);
126: String serializedInfo = serializeObject(info);
127: insertStatement.setString(1, serverUniqueId);
128: insertStatement.setString(2, deploymentId);
129: insertStatement.setString(3, serializedPrimaryKey);
130: insertStatement.setString(4, serializedInfo);
131: insertStatement.setLong(5, expiration.getTime());
132: insertStatement.setLong(6, intervalDuration);
133: int result = insertStatement.executeUpdate();
134: if (result != 1) {
135: throw new TimerStoreException(
136: "Could not insert!");
137: }
138: } finally {
139: insertStatement.close();
140: }
141: PreparedStatement identityStatement = c
142: .prepareStatement(identitySQL);
143: try {
144: ResultSet seqRS = identityStatement.executeQuery();
145: try {
146: seqRS.next();
147: id = seqRS.getLong(1);
148: } finally {
149: seqRS.close();
150: }
151: } finally {
152: identityStatement.close();
153: }
154: }
155: } catch (SQLException e) {
156: threwException = true;
157: throw new TimerStoreException(e);
158: } finally {
159: close(c, !threwException);
160: }
161:
162: TimerData timerData = new TimerData(id, timerService,
163: deploymentId, primaryKey, info, expiration,
164: intervalDuration);
165: return timerData;
166: }
167:
168: /**
169: * Used to restore a Timer that was cancelled, but the Transaction has been rolled back.
170: */
171: public void addTimerData(TimerData timerData)
172: throws TimerStoreException {
173: // TODO Need to verify how to handle this. Presumably, the Transaction rollback would "restore" this timer. So, no further action would be required.
174: // The MemoryTimerStore does require this capability...
175: }
176:
177: public void removeTimer(long timerId) {
178: boolean threwException = false;
179:
180: Connection c = null;
181: try {
182: c = getConnection();
183: PreparedStatement deleteStatement = c
184: .prepareStatement(deleteSQL);
185: try {
186: deleteStatement.setLong(1, timerId);
187: deleteStatement.execute();
188: } finally {
189: deleteStatement.close();
190: }
191: } catch (TimerStoreException e) {
192: log
193: .warning(
194: "Unable to remove get a database connection",
195: e);
196: } catch (SQLException e) {
197: threwException = true;
198: log.warning("Unable to remove timer data from database", e);
199: } finally {
200: close(c, !threwException);
201: }
202: }
203:
204: public Collection<TimerData> loadTimers(
205: EjbTimerServiceImpl timerService, String deploymentId)
206: throws TimerStoreException {
207: Collection<TimerData> timerDatas = new ArrayList<TimerData>();
208: boolean threwException = false;
209: Connection c = getConnection();
210: try {
211: PreparedStatement selectStatement = c
212: .prepareStatement(selectSQL);
213: selectStatement.setString(1, serverUniqueId);
214: selectStatement.setString(2, deploymentId);
215: try {
216: ResultSet taskRS = selectStatement.executeQuery();
217: try {
218: while (taskRS.next()) {
219: long id = taskRS.getLong(1);
220: String serizalizedUserId = taskRS.getString(2);
221: Object userId = deserializeObject(serizalizedUserId);
222: String serializedUserInfo = taskRS.getString(3);
223: Object userInfo = deserializeObject(serializedUserInfo);
224: long timeMillis = taskRS.getLong(4);
225: Date time = new Date(timeMillis);
226: long period = taskRS.getLong(5);
227:
228: TimerData timerData = new TimerData(id,
229: timerService, deploymentId, userId,
230: userInfo, time, period);
231: timerDatas.add(timerData);
232: }
233: } finally {
234: taskRS.close();
235: }
236: } finally {
237: selectStatement.close();
238: }
239: } catch (SQLException e) {
240: threwException = true;
241: throw new TimerStoreException(e);
242: } finally {
243: close(c, !threwException);
244: }
245: return timerDatas;
246: }
247:
248: public void updateIntervalTimer(TimerData timerData) {
249: boolean threwException = false;
250: Connection c = null;
251: try {
252: c = getConnection();
253: PreparedStatement updateStatement = c
254: .prepareStatement(fixedRateUpdateSQL);
255: try {
256: updateStatement.setLong(1, timerData.getExpiration()
257: .getTime());
258: updateStatement.setLong(2, timerData.getId());
259: updateStatement.execute();
260: } finally {
261: updateStatement.close();
262: }
263: } catch (TimerStoreException e) {
264: log
265: .warning(
266: "Unable to remove get a database connection",
267: e);
268: } catch (SQLException e) {
269: threwException = true;
270: log.warning("Unable to remove timer data from database", e);
271: } finally {
272: close(c, !threwException);
273: }
274: }
275:
276: private String serializeObject(Object object) {
277: try {
278: ByteArrayOutputStream baos = new ByteArrayOutputStream();
279: ObjectOutputStream out = new ObjectOutputStream(baos);
280: out.writeObject(object);
281: out.close();
282:
283: byte[] encoded = Base64.encodeBase64(baos.toByteArray());
284: return new String(encoded);
285: } catch (IOException e) {
286: throw new RuntimeException(e);
287: }
288: }
289:
290: private Object deserializeObject(String serializedValue) {
291: try {
292: byte[] bytes = Base64.decodeBase64(serializedValue
293: .getBytes());
294: ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
295: ObjectInputStream in = new ObjectInputStream(bais);
296: return in.readObject();
297: } catch (Exception e) {
298: throw new RuntimeException(e);
299: }
300: }
301:
302: private void execSQL(String sql) throws SQLException {
303: Connection connection = dataSource.getConnection();
304: try {
305: PreparedStatement updateStatement = connection
306: .prepareStatement(sql);
307: try {
308: updateStatement.execute();
309: } catch (SQLException e) {
310: //ignore... table already exists.
311: } finally {
312: updateStatement.close();
313: }
314: } finally {
315: connection.close();
316: }
317: }
318:
319: private Connection getConnection() throws TimerStoreException {
320: try {
321: return dataSource.getConnection();
322: } catch (Exception e) {
323: throw new TimerStoreException(e);
324: }
325: }
326:
327: private void close(Connection connection, boolean reportSqlException) {
328: if (connection != null) {
329: try {
330: connection.close();
331: } catch (Exception e) {
332: if (!reportSqlException) {
333: log.warning("Unable to close database connection",
334: e);
335: }
336: }
337: }
338: }
339: }
|