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 * @version $Revision: 1.1.1.1 $ 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 }