ActiveMQ example source code file (JournalPersistenceAdapter.java)

This example ActiveMQ source code file (JournalPersistenceAdapter.java) is included in the DevDaily.com "Java Source Code Warehouse" project. The intent of this project is to help you "Learn Java by Example" ^TM.

Java - ActiveMQ tags/keywords

exception, exception, failed, failed, futuretask, io, ioexception, ioexception, journal, journal, journalmessagestore, journaltopicmessagestore, journaltrace, persistenceadapter, recordlocation, threading, threads, util

The ActiveMQ JournalPersistenceAdapter.java source code

/**
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.activemq.store.journal;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.FutureTask;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.activeio.journal.InvalidRecordLocationException;
import org.apache.activeio.journal.Journal;
import org.apache.activeio.journal.JournalEventListener;
import org.apache.activeio.journal.RecordLocation;
import org.apache.activeio.packet.ByteArrayPacket;
import org.apache.activeio.packet.Packet;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.broker.BrokerServiceAware;
import org.apache.activemq.broker.ConnectionContext;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQTopic;
import org.apache.activemq.command.DataStructure;
import org.apache.activemq.command.JournalQueueAck;
import org.apache.activemq.command.JournalTopicAck;
import org.apache.activemq.command.JournalTrace;
import org.apache.activemq.command.JournalTransaction;
import org.apache.activemq.command.Message;
import org.apache.activemq.command.MessageAck;
import org.apache.activemq.command.ProducerId;
import org.apache.activemq.filter.NonCachedMessageEvaluationContext;
import org.apache.activemq.openwire.OpenWireFormat;
import org.apache.activemq.store.MessageStore;
import org.apache.activemq.store.PersistenceAdapter;
import org.apache.activemq.store.TopicMessageStore;
import org.apache.activemq.store.TransactionStore;
import org.apache.activemq.store.jdbc.JDBCPersistenceAdapter;
import org.apache.activemq.store.journal.JournalTransactionStore.Tx;
import org.apache.activemq.store.journal.JournalTransactionStore.TxOperation;
import org.apache.activemq.thread.Scheduler;
import org.apache.activemq.thread.Task;
import org.apache.activemq.thread.TaskRunner;
import org.apache.activemq.thread.TaskRunnerFactory;
import org.apache.activemq.usage.SystemUsage;
import org.apache.activemq.usage.Usage;
import org.apache.activemq.usage.UsageListener;
import org.apache.activemq.util.ByteSequence;
import org.apache.activemq.util.IOExceptionSupport;
import org.apache.activemq.wireformat.WireFormat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * An implementation of {@link PersistenceAdapter} designed for use with a
 * {@link Journal} and then check pointing asynchronously on a timeout with some
 * other long term persistent storage.
 * 
 * @org.apache.xbean.XBean
 * 
 */
public class JournalPersistenceAdapter implements PersistenceAdapter, JournalEventListener, UsageListener, BrokerServiceAware {

    private BrokerService brokerService;
	
    protected Scheduler scheduler;
    private static final Logger LOG = LoggerFactory.getLogger(JournalPersistenceAdapter.class);

    private Journal journal;
    private PersistenceAdapter longTermPersistence;

    private final WireFormat wireFormat = new OpenWireFormat();

    private final ConcurrentHashMap<ActiveMQQueue, JournalMessageStore> queues = new ConcurrentHashMap();
    private final ConcurrentHashMap<ActiveMQTopic, JournalTopicMessageStore> topics = new ConcurrentHashMap();

    private SystemUsage usageManager;
    private final long checkpointInterval = 1000 * 60 * 5;
    private long lastCheckpointRequest = System.currentTimeMillis();
    private long lastCleanup = System.currentTimeMillis();
    private int maxCheckpointWorkers = 10;
    private int maxCheckpointMessageAddSize = 1024 * 1024;

    private final JournalTransactionStore transactionStore = new JournalTransactionStore(this);
    private ThreadPoolExecutor checkpointExecutor;

    private TaskRunner checkpointTask;
    private CountDownLatch nextCheckpointCountDownLatch = new CountDownLatch(1);
    private boolean fullCheckPoint;

    private final AtomicBoolean started = new AtomicBoolean(false);

    private final Runnable periodicCheckpointTask = createPeriodicCheckpointTask();

    private TaskRunnerFactory taskRunnerFactory;

    public JournalPersistenceAdapter() {        
    }
    
    public JournalPersistenceAdapter(Journal journal, PersistenceAdapter longTermPersistence, TaskRunnerFactory taskRunnerFactory) throws IOException {
        setJournal(journal);
        setTaskRunnerFactory(taskRunnerFactory);
        setPersistenceAdapter(longTermPersistence);
    }

    public void setTaskRunnerFactory(TaskRunnerFactory taskRunnerFactory) {
        this.taskRunnerFactory = taskRunnerFactory;
    }

    public void setJournal(Journal journal) {
        this.journal = journal;
        journal.setJournalEventListener(this);
    }
    
    public void setPersistenceAdapter(PersistenceAdapter longTermPersistence) {
        this.longTermPersistence = longTermPersistence;
    }
    
    final Runnable createPeriodicCheckpointTask() {
        return new Runnable() {
            public void run() {
                long lastTime = 0;
                synchronized (this) {
                    lastTime = lastCheckpointRequest;
                }
                if (System.currentTimeMillis() > lastTime + checkpointInterval) {
                    checkpoint(false, true);
                }
            }
        };
    }

    /**
     * @param usageManager The UsageManager that is controlling the
     *                destination's memory usage.
     */
    public void setUsageManager(SystemUsage usageManager) {
        this.usageManager = usageManager;
        longTermPersistence.setUsageManager(usageManager);
    }

    public Set<ActiveMQDestination> getDestinations() {
        Set<ActiveMQDestination> destinations = new HashSet(longTermPersistence.getDestinations());
        destinations.addAll(queues.keySet());
        destinations.addAll(topics.keySet());
        return destinations;
    }

    private MessageStore createMessageStore(ActiveMQDestination destination) throws IOException {
        if (destination.isQueue()) {
            return createQueueMessageStore((ActiveMQQueue)destination);
        } else {
            return createTopicMessageStore((ActiveMQTopic)destination);
        }
    }

    public MessageStore createQueueMessageStore(ActiveMQQueue destination) throws IOException {
        JournalMessageStore store = queues.get(destination);
        if (store == null) {
            MessageStore checkpointStore = longTermPersistence.createQueueMessageStore(destination);
            store = new JournalMessageStore(this, checkpointStore, destination);
            queues.put(destination, store);
        }
        return store;
    }

    public TopicMessageStore createTopicMessageStore(ActiveMQTopic destinationName) throws IOException {
        JournalTopicMessageStore store = topics.get(destinationName);
        if (store == null) {
            TopicMessageStore checkpointStore = longTermPersistence.createTopicMessageStore(destinationName);
            store = new JournalTopicMessageStore(this, checkpointStore, destinationName);
            topics.put(destinationName, store);
        }
        return store;
    }

    /**
     * Cleanup method to remove any state associated with the given destination
     *
     * @param destination Destination to forget
     */
    public void removeQueueMessageStore(ActiveMQQueue destination) {
        queues.remove(destination);
    }

    /**
     * Cleanup method to remove any state associated with the given destination
     *
     * @param destination Destination to forget
     */
    public void removeTopicMessageStore(ActiveMQTopic destination) {
        topics.remove(destination);
    }

    public TransactionStore createTransactionStore() throws IOException {
        return transactionStore;
    }

    public long getLastMessageBrokerSequenceId() throws IOException {
        return longTermPersistence.getLastMessageBrokerSequenceId();
    }

    public void beginTransaction(ConnectionContext context) throws IOException {
        longTermPersistence.beginTransaction(context);
    }

    public void commitTransaction(ConnectionContext context) throws IOException {
        longTermPersistence.commitTransaction(context);
    }

    public void rollbackTransaction(ConnectionContext context) throws IOException {
        longTermPersistence.rollbackTransaction(context);
    }

    public synchronized void start() throws Exception {
        if (!started.compareAndSet(false, true)) {
            return;
        }

        checkpointTask = taskRunnerFactory.createTaskRunner(new Task() {
            public boolean iterate() {
                return doCheckpoint();
            }
        }, "ActiveMQ Journal Checkpoint Worker");

        checkpointExecutor = new ThreadPoolExecutor(maxCheckpointWorkers, maxCheckpointWorkers, 30, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(), new ThreadFactory() {
            public Thread newThread(Runnable runable) {
                Thread t = new Thread(runable, "Journal checkpoint worker");
                t.setPriority(7);
                return t;
            }
        });
        // checkpointExecutor.allowCoreThreadTimeOut(true);

        this.usageManager.getMemoryUsage().addUsageListener(this);

        if (longTermPersistence instanceof JDBCPersistenceAdapter) {
            // Disabled periodic clean up as it deadlocks with the checkpoint
            // operations.
            ((JDBCPersistenceAdapter)longTermPersistence).setCleanupPeriod(0);
        }

        longTermPersistence.start();
        createTransactionStore();
        recover();

        // Do a checkpoint periodically.
        this.scheduler = new Scheduler("Journal Scheduler");
        this.scheduler.start();
        this.scheduler.executePeriodically(periodicCheckpointTask, checkpointInterval / 10);

    }

    public void stop() throws Exception {

        this.usageManager.getMemoryUsage().removeUsageListener(this);
        if (!started.compareAndSet(true, false)) {
            return;
        }

        this.scheduler.cancel(periodicCheckpointTask);
        this.scheduler.stop();

        // Take one final checkpoint and stop checkpoint processing.
        checkpoint(true, true);
        checkpointTask.shutdown();
        checkpointExecutor.shutdown();

        queues.clear();
        topics.clear();

        IOException firstException = null;
        try {
            journal.close();
        } catch (Exception e) {
            firstException = IOExceptionSupport.create("Failed to close journals: " + e, e);
        }
        longTermPersistence.stop();

        if (firstException != null) {
            throw firstException;
        }
    }

    // Properties
    // -------------------------------------------------------------------------
    public PersistenceAdapter getLongTermPersistence() {
        return longTermPersistence;
    }

    /**
     * @return Returns the wireFormat.
     */
    public WireFormat getWireFormat() {
        return wireFormat;
    }

    // Implementation methods
    // -------------------------------------------------------------------------

    /**
     * The Journal give us a call back so that we can move old data out of the
     * journal. Taking a checkpoint does this for us.
     * 
     * @see org.apache.activemq.journal.JournalEventListener#overflowNotification(org.apache.activemq.journal.RecordLocation)
     */
    public void overflowNotification(RecordLocation safeLocation) {
        checkpoint(false, true);
    }

    /**
     * When we checkpoint we move all the journalled data to long term storage.
     * 
     */
    public void checkpoint(boolean sync, boolean fullCheckpoint) {
        try {
            if (journal == null) {
                throw new IllegalStateException("Journal is closed.");
            }

            long now = System.currentTimeMillis();
            CountDownLatch latch = null;
            synchronized (this) {
                latch = nextCheckpointCountDownLatch;
                lastCheckpointRequest = now;
                if (fullCheckpoint) {
                    this.fullCheckPoint = true;
                }
            }

            checkpointTask.wakeup();

            if (sync) {
                LOG.debug("Waking for checkpoint to complete.");
                latch.await();
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            LOG.warn("Request to start checkpoint failed: " + e, e);
        }
    }

    public void checkpoint(boolean sync) {
        checkpoint(sync, sync);
    }

    /**
     * This does the actual checkpoint.
     * 
     * @return
     */
    public boolean doCheckpoint() {
        CountDownLatch latch = null;
        boolean fullCheckpoint;
        synchronized (this) {
            latch = nextCheckpointCountDownLatch;
            nextCheckpointCountDownLatch = new CountDownLatch(1);
            fullCheckpoint = this.fullCheckPoint;
            this.fullCheckPoint = false;
        }
        try {

            LOG.debug("Checkpoint started.");
            RecordLocation newMark = null;

            ArrayList<FutureTask futureTasks = new ArrayList>(queues.size() + topics.size());

            //
            // We do many partial checkpoints (fullCheckpoint==false) to move
            // topic messages
            // to long term store as soon as possible.
            // 
            // We want to avoid doing that for queue messages since removes the
            // come in the same
            // checkpoint cycle will nullify the previous message add.
            // Therefore, we only
            // checkpoint queues on the fullCheckpoint cycles.
            //
            if (fullCheckpoint) {
                Iterator<JournalMessageStore> iterator = queues.values().iterator();
                while (iterator.hasNext()) {
                    try {
                        final JournalMessageStore ms = iterator.next();
                        FutureTask<RecordLocation> task = new FutureTask(new Callable() {
                            public RecordLocation call() throws Exception {
                                return ms.checkpoint();
                            }
                        });
                        futureTasks.add(task);
                        checkpointExecutor.execute(task);
                    } catch (Exception e) {
                        LOG.error("Failed to checkpoint a message store: " + e, e);
                    }
                }
            }

            Iterator<JournalTopicMessageStore> iterator = topics.values().iterator();
            while (iterator.hasNext()) {
                try {
                    final JournalTopicMessageStore ms = iterator.next();
                    FutureTask<RecordLocation> task = new FutureTask(new Callable() {
                        public RecordLocation call() throws Exception {
                            return ms.checkpoint();
                        }
                    });
                    futureTasks.add(task);
                    checkpointExecutor.execute(task);
                } catch (Exception e) {
                    LOG.error("Failed to checkpoint a message store: " + e, e);
                }
            }

            try {
                for (Iterator<FutureTask iter = futureTasks.iterator(); iter.hasNext();) {
                    FutureTask<RecordLocation> ft = iter.next();
                    RecordLocation mark = ft.get();
                    // We only set a newMark on full checkpoints.
                    if (fullCheckpoint) {
                        if (mark != null && (newMark == null || newMark.compareTo(mark) < 0)) {
                            newMark = mark;
                        }
                    }
                }
            } catch (Throwable e) {
                LOG.error("Failed to checkpoint a message store: " + e, e);
            }

            if (fullCheckpoint) {
                try {
                    if (newMark != null) {
                        LOG.debug("Marking journal at: " + newMark);
                        journal.setMark(newMark, true);
                    }
                } catch (Exception e) {
                    LOG.error("Failed to mark the Journal: " + e, e);
                }

                if (longTermPersistence instanceof JDBCPersistenceAdapter) {
                    // We may be check pointing more often than the
                    // checkpointInterval if under high use
                    // But we don't want to clean up the db that often.
                    long now = System.currentTimeMillis();
                    if (now > lastCleanup + checkpointInterval) {
                        lastCleanup = now;
                        ((JDBCPersistenceAdapter)longTermPersistence).cleanup();
                    }
                }
            }

            LOG.debug("Checkpoint done.");
        } finally {
            latch.countDown();
        }
        synchronized (this) {
            return this.fullCheckPoint;
        }

    }

    /**
     * @param location
     * @return
     * @throws IOException
     */
    public DataStructure readCommand(RecordLocation location) throws IOException {
        try {
            Packet packet = journal.read(location);
            return (DataStructure)wireFormat.unmarshal(toByteSequence(packet));
        } catch (InvalidRecordLocationException e) {
            throw createReadException(location, e);
        } catch (IOException e) {
            throw createReadException(location, e);
        }
    }

    /**
     * Move all the messages that were in the journal into long term storage. We
     * just replay and do a checkpoint.
     * 
     * @throws IOException
     * @throws IOException
     * @throws InvalidRecordLocationException
     * @throws IllegalStateException
     */
    private void recover() throws IllegalStateException, InvalidRecordLocationException, IOException, IOException {

        RecordLocation pos = null;
        int transactionCounter = 0;

        LOG.info("Journal Recovery Started from: " + journal);
        ConnectionContext context = new ConnectionContext(new NonCachedMessageEvaluationContext());

        // While we have records in the journal.
        while ((pos = journal.getNextRecordLocation(pos)) != null) {
            Packet data = journal.read(pos);
            DataStructure c = (DataStructure)wireFormat.unmarshal(toByteSequence(data));

            if (c instanceof Message) {
                Message message = (Message)c;
                JournalMessageStore store = (JournalMessageStore)createMessageStore(message.getDestination());
                if (message.isInTransaction()) {
                    transactionStore.addMessage(store, message, pos);
                } else {
                    store.replayAddMessage(context, message);
                    transactionCounter++;
                }
            } else {
                switch (c.getDataStructureType()) {
                case JournalQueueAck.DATA_STRUCTURE_TYPE: {
                    JournalQueueAck command = (JournalQueueAck)c;
                    JournalMessageStore store = (JournalMessageStore)createMessageStore(command.getDestination());
                    if (command.getMessageAck().isInTransaction()) {
                        transactionStore.removeMessage(store, command.getMessageAck(), pos);
                    } else {
                        store.replayRemoveMessage(context, command.getMessageAck());
                        transactionCounter++;
                    }
                }
                    break;
                case JournalTopicAck.DATA_STRUCTURE_TYPE: {
                    JournalTopicAck command = (JournalTopicAck)c;
                    JournalTopicMessageStore store = (JournalTopicMessageStore)createMessageStore(command.getDestination());
                    if (command.getTransactionId() != null) {
                        transactionStore.acknowledge(store, command, pos);
                    } else {
                        store.replayAcknowledge(context, command.getClientId(), command.getSubscritionName(), command.getMessageId());
                        transactionCounter++;
                    }
                }
                    break;
                case JournalTransaction.DATA_STRUCTURE_TYPE: {
                    JournalTransaction command = (JournalTransaction)c;
                    try {
                        // Try to replay the packet.
                        switch (command.getType()) {
                        case JournalTransaction.XA_PREPARE:
                            transactionStore.replayPrepare(command.getTransactionId());
                            break;
                        case JournalTransaction.XA_COMMIT:
                        case JournalTransaction.LOCAL_COMMIT:
                            Tx tx = transactionStore.replayCommit(command.getTransactionId(), command.getWasPrepared());
                            if (tx == null) {
                                break; // We may be trying to replay a commit
                            }
                            // that
                            // was already committed.

                            // Replay the committed operations.
                            tx.getOperations();
                            for (Iterator iter = tx.getOperations().iterator(); iter.hasNext();) {
                                TxOperation op = (TxOperation)iter.next();
                                if (op.operationType == TxOperation.ADD_OPERATION_TYPE) {
                                    op.store.replayAddMessage(context, (Message)op.data);
                                }
                                if (op.operationType == TxOperation.REMOVE_OPERATION_TYPE) {
                                    op.store.replayRemoveMessage(context, (MessageAck)op.data);
                                }
                                if (op.operationType == TxOperation.ACK_OPERATION_TYPE) {
                                    JournalTopicAck ack = (JournalTopicAck)op.data;
                                    ((JournalTopicMessageStore)op.store).replayAcknowledge(context, ack.getClientId(), ack.getSubscritionName(), ack.getMessageId());
                                }
                            }
                            transactionCounter++;
                            break;
                        case JournalTransaction.LOCAL_ROLLBACK:
                        case JournalTransaction.XA_ROLLBACK:
                            transactionStore.replayRollback(command.getTransactionId());
                            break;
                        default:
                            throw new IOException("Invalid journal command type: " + command.getType());
                        }
                    } catch (IOException e) {
                        LOG.error("Recovery Failure: Could not replay: " + c + ", reason: " + e, e);
                    }
                }
                    break;
                case JournalTrace.DATA_STRUCTURE_TYPE:
                    JournalTrace trace = (JournalTrace)c;
                    LOG.debug("TRACE Entry: " + trace.getMessage());
                    break;
                default:
                    LOG.error("Unknown type of record in transaction log which will be discarded: " + c);
                }
            }
        }

        RecordLocation location = writeTraceMessage("RECOVERED", true);
        journal.setMark(location, true);

        LOG.info("Journal Recovered: " + transactionCounter + " message(s) in transactions recovered.");
    }

    private IOException createReadException(RecordLocation location, Exception e) {
        return IOExceptionSupport.create("Failed to read to journal for: " + location + ". Reason: " + e, e);
    }

    protected IOException createWriteException(DataStructure packet, Exception e) {
        return IOExceptionSupport.create("Failed to write to journal for: " + packet + ". Reason: " + e, e);
    }

    protected IOException createWriteException(String command, Exception e) {
        return IOExceptionSupport.create("Failed to write to journal for command: " + command + ". Reason: " + e, e);
    }

    protected IOException createRecoveryFailedException(Exception e) {
        return IOExceptionSupport.create("Failed to recover from journal. Reason: " + e, e);
    }

    /**
     * @param command
     * @param sync
     * @return
     * @throws IOException
     */
    public RecordLocation writeCommand(DataStructure command, boolean sync) throws IOException {
        if (started.get()) {
            try {
        	    return journal.write(toPacket(wireFormat.marshal(command)), sync);
            } catch (IOException ioe) {
        	    LOG.error("Cannot write to the journal", ioe);
        	    brokerService.handleIOException(ioe);
        	    throw ioe;
            }
        }
        throw new IOException("closed");
    }

    private RecordLocation writeTraceMessage(String message, boolean sync) throws IOException {
        JournalTrace trace = new JournalTrace();
        trace.setMessage(message);
        return writeCommand(trace, sync);
    }

    public void onUsageChanged(Usage usage, int oldPercentUsage, int newPercentUsage) {
        newPercentUsage = (newPercentUsage / 10) * 10;
        oldPercentUsage = (oldPercentUsage / 10) * 10;
        if (newPercentUsage >= 70 && oldPercentUsage < newPercentUsage) {
            boolean sync = newPercentUsage >= 90;
            checkpoint(sync, true);
        }
    }

    public JournalTransactionStore getTransactionStore() {
        return transactionStore;
    }

    public void deleteAllMessages() throws IOException {
        try {
            JournalTrace trace = new JournalTrace();
            trace.setMessage("DELETED");
            RecordLocation location = journal.write(toPacket(wireFormat.marshal(trace)), false);
            journal.setMark(location, true);
            LOG.info("Journal deleted: ");
        } catch (IOException e) {
            throw e;
        } catch (Throwable e) {
            throw IOExceptionSupport.create(e);
        }
        longTermPersistence.deleteAllMessages();
    }

    public SystemUsage getUsageManager() {
        return usageManager;
    }

    public int getMaxCheckpointMessageAddSize() {
        return maxCheckpointMessageAddSize;
    }

    public void setMaxCheckpointMessageAddSize(int maxCheckpointMessageAddSize) {
        this.maxCheckpointMessageAddSize = maxCheckpointMessageAddSize;
    }

    public int getMaxCheckpointWorkers() {
        return maxCheckpointWorkers;
    }

    public void setMaxCheckpointWorkers(int maxCheckpointWorkers) {
        this.maxCheckpointWorkers = maxCheckpointWorkers;
    }

    public boolean isUseExternalMessageReferences() {
        return false;
    }

    public void setUseExternalMessageReferences(boolean enable) {
        if (enable) {
            throw new IllegalArgumentException("The journal does not support message references.");
        }
    }

    public Packet toPacket(ByteSequence sequence) {
        return new ByteArrayPacket(new org.apache.activeio.packet.ByteSequence(sequence.data, sequence.offset, sequence.length));
    }

    public ByteSequence toByteSequence(Packet packet) {
        org.apache.activeio.packet.ByteSequence sequence = packet.asByteSequence();
        return new ByteSequence(sequence.getData(), sequence.getOffset(), sequence.getLength());
    }

    public void setBrokerName(String brokerName) {
        longTermPersistence.setBrokerName(brokerName);
    }

    @Override
    public String toString() {
        return "JournalPersistenceAdapator(" + longTermPersistence + ")";
    }

    public void setDirectory(File dir) {
    }
    
    public long size(){
        return 0;
    }

    public void setBrokerService(BrokerService brokerService) {
        this.brokerService = brokerService;
        PersistenceAdapter pa = getLongTermPersistence();
        if( pa instanceof BrokerServiceAware ) {
            ((BrokerServiceAware)pa).setBrokerService(brokerService);
        }
    }

    public long getLastProducerSequenceId(ProducerId id) {
        return -1;
    }

}