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 */
017 package org.apache.activemq.kaha.impl.async;
018
019 import java.io.IOException;
020
021 import org.apache.activeio.journal.InvalidRecordLocationException;
022 import org.apache.activeio.journal.Journal;
023 import org.apache.activeio.journal.JournalEventListener;
024 import org.apache.activeio.journal.RecordLocation;
025 import org.apache.activeio.packet.ByteArrayPacket;
026 import org.apache.activeio.packet.Packet;
027 import org.apache.activemq.util.ByteSequence;
028
029 /**
030 * Provides a Journal Facade to the DataManager.
031 *
032 *
033 */
034 public final class JournalFacade implements Journal {
035
036 private final AsyncDataManager dataManager;
037
038 public static class RecordLocationFacade implements RecordLocation {
039 private final Location location;
040
041 public RecordLocationFacade(Location location) {
042 this.location = location;
043 }
044
045 public Location getLocation() {
046 return location;
047 }
048
049 public int compareTo(Object o) {
050 RecordLocationFacade rlf = (RecordLocationFacade)o;
051 int rc = location.compareTo(rlf.location);
052 return rc;
053 }
054 }
055
056 public JournalFacade(AsyncDataManager dataManager) {
057 this.dataManager = dataManager;
058 }
059
060 private static RecordLocation convertToRecordLocation(Location location) {
061 if (location == null) {
062 return null;
063 }
064 return new RecordLocationFacade(location);
065 }
066
067 private static Location convertFromRecordLocation(RecordLocation location) {
068
069 if (location == null) {
070 return null;
071 }
072
073 return ((RecordLocationFacade)location).getLocation();
074 }
075
076 public void close() throws IOException {
077 dataManager.close();
078 }
079
080 public RecordLocation getMark() throws IllegalStateException {
081 return convertToRecordLocation(dataManager.getMark());
082 }
083
084 public RecordLocation getNextRecordLocation(RecordLocation location) throws InvalidRecordLocationException, IOException, IllegalStateException {
085 return convertToRecordLocation(dataManager.getNextLocation(convertFromRecordLocation(location)));
086 }
087
088 public Packet read(RecordLocation location) throws InvalidRecordLocationException, IOException, IllegalStateException {
089 ByteSequence rc = dataManager.read(convertFromRecordLocation(location));
090 if (rc == null) {
091 return null;
092 }
093 return new ByteArrayPacket(rc.getData(), rc.getOffset(), rc.getLength());
094 }
095
096 public void setJournalEventListener(JournalEventListener listener) throws IllegalStateException {
097 }
098
099 public void setMark(RecordLocation location, boolean sync) throws InvalidRecordLocationException, IOException, IllegalStateException {
100 dataManager.setMark(convertFromRecordLocation(location), sync);
101 }
102
103 public RecordLocation write(Packet packet, boolean sync) throws IOException, IllegalStateException {
104 org.apache.activeio.packet.ByteSequence data = packet.asByteSequence();
105 ByteSequence sequence = new ByteSequence(data.getData(), data.getOffset(), data.getLength());
106 return convertToRecordLocation(dataManager.write(sequence, sync));
107 }
108
109 public RecordLocation write(Packet packet, Runnable onComplete) throws IOException, IllegalStateException {
110 org.apache.activeio.packet.ByteSequence data = packet.asByteSequence();
111 ByteSequence sequence = new ByteSequence(data.getData(), data.getOffset(), data.getLength());
112 return convertToRecordLocation(dataManager.write(sequence, onComplete));
113 }
114
115 }