You are viewing a plain text version of this content. The canonical link for it is here.
Posted to server-dev@james.apache.org by ie...@apache.org on 2011/09/04 09:31:43 UTC
svn commit: r1164966 [1/5] - in
/james/mailbox/tags/apache-james-mailbox-0.3: ./ hbase/ hbase/src/
hbase/src/main/ hbase/src/main/config/ hbase/src/main/java/
hbase/src/main/java/org/ hbase/src/main/java/org/apache/
hbase/src/main/java/org/apache/james...
Author: ieugen
Date: Sun Sep 4 07:31:41 2011
New Revision: 1164966
URL: http://svn.apache.org/viewvc?rev=1164966&view=rev
Log:
Added HBase mailbox implementation to 0.3 branch.
Please see https://issues.apache.org/jira/browse/MAILBOX-44 for details.
Added:
james/mailbox/tags/apache-james-mailbox-0.3/hbase/
james/mailbox/tags/apache-james-mailbox-0.3/hbase/pom.xml
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/config/
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/config/hbase-site.xml
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/FlagConvertor.java
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/HBaseMailboxManager.java
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/HBaseMailboxSessionMapperFactory.java
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/HBaseMessageManager.java
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/HBaseNames.java
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/HBaseNonTransactionalMapper.java
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/HBaseUtils.java
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/PropertyConvertor.java
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/io/
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/io/ChunkInputStream.java
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/io/ChunkOutputStream.java
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/mail/
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/mail/HBaseMailboxMapper.java
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/mail/HBaseMessage.java
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/mail/HBaseMessageMapper.java
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/mail/HBaseModSeqProvider.java
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/mail/HBaseUidProvider.java
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/mail/model/
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/mail/model/HBaseMailbox.java
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/user/
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/user/HBaseSubscriptionMapper.java
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/reporting-site/
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/reporting-site/site.xml
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/site/
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/site/James HBase Mailbox Schema.mm
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/site/James HBase Mailbox Schema.svg
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/test/
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/test/java/
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/test/java/org/
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/test/java/org/apache/
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/test/java/org/apache/james/
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/test/java/org/apache/james/mailbox/
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/test/java/org/apache/james/mailbox/hbase/
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/test/java/org/apache/james/mailbox/hbase/HBaseClusterSingleton.java
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/test/java/org/apache/james/mailbox/hbase/HBaseMailboxManagerTest.java
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/test/java/org/apache/james/mailbox/hbase/HBaseMailboxSessionMapperFactoryTest.java
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/test/java/org/apache/james/mailbox/hbase/HBaseUtilsTest.java
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/test/java/org/apache/james/mailbox/hbase/mail/
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/test/java/org/apache/james/mailbox/hbase/mail/HBaseMailboxMapperTest.java
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/test/java/org/apache/james/mailbox/hbase/mail/HBaseMessageMapperTest.java
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/test/java/org/apache/james/mailbox/hbase/mail/HBaseUidAndModSeqProviderTest.java
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/test/java/org/apache/james/mailbox/hbase/mail/model/
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/test/java/org/apache/james/mailbox/hbase/mail/model/HBaseMailboxTest.java
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/test/java/org/apache/james/mailbox/hbase/user/
james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/test/java/org/apache/james/mailbox/hbase/user/HBaseSubscriptionMapperTest.java
Modified:
james/mailbox/tags/apache-james-mailbox-0.3/pom.xml
Added: james/mailbox/tags/apache-james-mailbox-0.3/hbase/pom.xml
URL: http://svn.apache.org/viewvc/james/mailbox/tags/apache-james-mailbox-0.3/hbase/pom.xml?rev=1164966&view=auto
==============================================================================
--- james/mailbox/tags/apache-james-mailbox-0.3/hbase/pom.xml (added)
+++ james/mailbox/tags/apache-james-mailbox-0.3/hbase/pom.xml Sun Sep 4 07:31:41 2011
@@ -0,0 +1,193 @@
+<?xml version="1.0" encoding="ISO-8859-15"?>
+<!--
+ 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+ <parent>
+ <artifactId>apache-james-mailbox</artifactId>
+ <groupId>org.apache.james</groupId>
+ <version>0.3</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+ <groupId>org.apache.james</groupId>
+ <artifactId>apache-james-mailbox-hbase</artifactId>
+ <name>Apache James HBase Mailbox</name>
+ <description>Apache James Mailbox implementation over HBase</description>
+
+ <properties>
+ <hbase.version>0.90.3</hbase.version>
+ <hadoop.version>0.20-append-r1057313</hadoop.version>
+ <commons-io.version>2.0.1</commons-io.version>
+ <junit.version>4.8.2</junit.version>
+ <log4j.version>1.2.16</log4j.version>
+ <slf4j.version>1.6.1</slf4j.version>
+ <target.jdk>1.6</target.jdk>
+ </properties>
+
+ <repositories>
+ <repository>
+ <id>hadoop-non-releases</id>
+ <name>Hadoop non-releases</name>
+ <url>http://people.apache.org/~rawson/repo/</url>
+ <snapshots>
+ <enabled>false</enabled>
+ </snapshots>
+ <releases>
+ <enabled>true</enabled>
+ </releases>
+ </repository>
+ </repositories>
+
+ <dependencyManagement>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase</artifactId>
+ <version>0.90.3</version>
+ </dependency>
+ <dependency>
+ <groupId>commons-io</groupId>
+ <artifactId>commons-io</artifactId>
+ <version>${commons-io.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ <version>1.6.1</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j</artifactId>
+ <version>${slf4j.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j</artifactId>
+ <version>${slf4j.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>log4j</groupId>
+ <artifactId>log4j</artifactId>
+ <version>${log4j.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <version>${junit.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase</artifactId>
+ <version>${hbase.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-test</artifactId>
+ <version>0.20-append-r1057313</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+ </dependencyManagement>
+
+
+ <dependencies>
+ <dependency>
+ <groupId>${javax.mail.groupId}</groupId>
+ <artifactId>${javax.mail.artifactId}</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit-dep</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>commons-io</groupId>
+ <artifactId>commons-io</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-simple</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>apache-james-mailbox-api</artifactId>
+ <version>0.3</version>
+ </dependency>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>apache-james-mailbox-store</artifactId>
+ <version>0.3</version>
+ </dependency>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>apache-james-mailbox-api</artifactId>
+ <version>0.3</version>
+ <scope>test</scope>
+ <type>test-jar</type>
+ </dependency>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>apache-james-mailbox-store</artifactId>
+ <version>0.3</version>
+ <scope>test</scope>
+ <type>test-jar</type>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase</artifactId>
+ <exclusions>
+ <exclusion>
+ <artifactId>hadoop-core</artifactId>
+ <groupId>org.apache.hadoop</groupId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase</artifactId>
+ <type>test-jar</type>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-test</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-core</artifactId>
+ <version>0.20-append-r1056497</version>
+ </dependency>
+ <dependency>
+ <groupId>jaxme</groupId>
+ <artifactId>jaxme2</artifactId>
+ <version>0.5.1</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+</project>
Added: james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/config/hbase-site.xml
URL: http://svn.apache.org/viewvc/james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/config/hbase-site.xml?rev=1164966&view=auto
==============================================================================
--- james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/config/hbase-site.xml (added)
+++ james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/config/hbase-site.xml Sun Sep 4 07:31:41 2011
@@ -0,0 +1,49 @@
+<?xml version="1.0"?>
+<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
+<!--
+/**
+ * Copyright 2011 The Apache Software Foundation
+ *
+ * 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.
+ */
+-->
+<configuration>
+<!--
+ This config file must be on the application classpath and will tell our application
+ where to find the HBase cluster.
+-->
+ <property>
+ <name>hbase.rootdir</name>
+ <value>file:///home/ieugen/contracte/gsoc2011/hbaseROOT/hbase</value>
+ </property>
+
+ <property>
+ <name>hbase.master.port</name>
+ <value>60000</value>
+ </property>
+
+ <property>
+ <name>hbase.regionserver.info.port</name>
+ <value>6030</value>
+ </property>
+
+ <property>
+ <name>hbase.regionserver.info.bindAddress</name>
+ <value>0.0.0.0</value>
+ </property>
+
+</configuration>
Added: james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/FlagConvertor.java
URL: http://svn.apache.org/viewvc/james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/FlagConvertor.java?rev=1164966&view=auto
==============================================================================
--- james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/FlagConvertor.java (added)
+++ james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/FlagConvertor.java Sun Sep 4 07:31:41 2011
@@ -0,0 +1,133 @@
+/****************************************************************
+ * 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.james.mailbox.hbase;
+
+import javax.mail.Flags;
+import javax.mail.Flags.Flag;
+import org.apache.hadoop.hbase.util.Bytes;
+
+/**
+ * Class used for converting message flags to and from byte arrays for use as
+ * HBase column qualifiers.
+ */
+public class FlagConvertor {
+
+ public static final String PREFIX_SFLAGS = "sf:";
+ public static final byte[] PREFIX_SFLAGS_B = Bytes.toBytes(PREFIX_SFLAGS);
+ public static final String PREFIX_UFLAGS = "uf:";
+ public static final byte[] PREFIX_UFLAGS_B = Bytes.toBytes(PREFIX_UFLAGS);
+ /*TODO: find a way to store all flags as a single byte (HBase bitewise operations).
+ * this will be efficient also because we will not store the column names.
+ */
+ public static final byte[] FLAGS_ANSWERED = Bytes.toBytes("sf:A");
+ public static final byte[] FLAGS_DELETED = Bytes.toBytes("sf:DE");
+ public static final byte[] FLAGS_DRAFT = Bytes.toBytes("sf:DR");
+ public static final byte[] FLAGS_FLAGGED = Bytes.toBytes("sf:F");
+ public static final byte[] FLAGS_RECENT = Bytes.toBytes("sf:R");
+ public static final byte[] FLAGS_SEEN = Bytes.toBytes("sf:S");
+ public static final byte[] FLAGS_USER = Bytes.toBytes("sf:U");
+
+ /**
+ * Converts a {@link javax.mail.Flags.Flag} to a byte array representation
+ * used for storing in HBase (as a column qualifier).
+ * @param flag
+ * @return a byte representation of the flag.
+ *
+ * Throws RuntimeException if Flag is not recognized.
+ */
+ public static byte[] systemFlagToBytes(Flags.Flag flag) {
+ if (flag.equals(Flag.ANSWERED)) {
+ return FLAGS_ANSWERED;
+ }
+ if (flag.equals(Flag.DELETED)) {
+ return FLAGS_DELETED;
+ }
+ if (flag.equals(Flag.DRAFT)) {
+ return FLAGS_DRAFT;
+ }
+ if (flag.equals(Flag.FLAGGED)) {
+ return FLAGS_FLAGGED;
+ }
+ if (flag.equals(Flag.RECENT)) {
+ return FLAGS_RECENT;
+ }
+ if (flag.equals(Flag.SEEN)) {
+ return FLAGS_SEEN;
+ }
+ if (flag.equals(Flag.USER)) {
+ return FLAGS_USER;
+ }
+ throw new RuntimeException("Invalid Flag supplied");
+ }
+
+ /**
+ * Returns a {@link javax.mail.Flags.Flag} coresponding to the supplyed
+ * byte array.
+ * @param bytes byte array representation
+ * @return one of {@link javax.mail.Flags.Flag}
+ * @throws RuntimeException if the byte array does not match a
+ * suitable represetnation.
+ */
+ public static Flag systemFlagFromBytes(byte[] bytes) {
+ if (Bytes.equals(bytes, FLAGS_ANSWERED)) {
+ return Flag.ANSWERED;
+ }
+ if (Bytes.equals(bytes, FLAGS_DELETED)) {
+ return Flag.DELETED;
+ }
+ if (Bytes.equals(bytes, FLAGS_DRAFT)) {
+ return Flag.DRAFT;
+ }
+ if (Bytes.equals(bytes, FLAGS_FLAGGED)) {
+ return Flag.FLAGGED;
+ }
+ if (Bytes.equals(bytes, FLAGS_RECENT)) {
+ return Flag.RECENT;
+ }
+ if (Bytes.equals(bytes, FLAGS_SEEN)) {
+ return Flag.SEEN;
+ }
+ if (Bytes.equals(bytes, FLAGS_USER)) {
+ return Flag.USER;
+ }
+ throw new RuntimeException("This is not a recognized system flag: " + Bytes.toString(bytes));
+ }
+
+ /**
+ * Converts a user flag to a byte array for use as a HBase column qualifier.
+ * @param flag user flag to convert
+ * @return a byte array representation of the user flag
+ */
+ public static byte[] userFlagToBytes(String flag) {
+ return Bytes.toBytes(PREFIX_UFLAGS + flag);
+ }
+
+ /**
+ * Converts a byte array to a user flag.
+ * @param bytes the user flag byte representation
+ * @return a {@link String} representaion of the user flag
+ * @throws RuntimeException if the user flag prefix is not found.
+ */
+ public static String userFlagFromBytes(byte[] bytes) {
+ if (Bytes.startsWith(bytes, PREFIX_UFLAGS_B)) {
+ return Bytes.toString(bytes, PREFIX_UFLAGS_B.length, bytes.length - PREFIX_UFLAGS_B.length);
+ }
+ throw new RuntimeException("This is not a user flag representation: " + Bytes.toString(bytes));
+ }
+}
Added: james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/HBaseMailboxManager.java
URL: http://svn.apache.org/viewvc/james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/HBaseMailboxManager.java?rev=1164966&view=auto
==============================================================================
--- james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/HBaseMailboxManager.java (added)
+++ james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/HBaseMailboxManager.java Sun Sep 4 07:31:41 2011
@@ -0,0 +1,85 @@
+/****************************************************************
+ * 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.james.mailbox.hbase;
+
+import java.util.UUID;
+import org.apache.james.mailbox.hbase.mail.model.HBaseMailbox;
+import org.apache.james.mailbox.MailboxPathLocker;
+import org.apache.james.mailbox.hbase.mail.HBaseMailboxMapper;
+import org.apache.james.mailbox.MailboxException;
+import org.apache.james.mailbox.MailboxPath;
+import org.apache.james.mailbox.MailboxSession;
+import org.apache.james.mailbox.store.Authenticator;
+import org.apache.james.mailbox.store.JVMMailboxPathLocker;
+import org.apache.james.mailbox.store.StoreMailboxManager;
+import org.apache.james.mailbox.store.StoreMessageManager;
+import org.apache.james.mailbox.store.mail.model.Mailbox;
+import org.apache.james.mailbox.store.transaction.TransactionalMapper;
+
+/**
+ * HBase implementation of {@link StoreMailboxManager}
+ *
+ */
+public class HBaseMailboxManager extends StoreMailboxManager<UUID> {
+
+ public HBaseMailboxManager(HBaseMailboxSessionMapperFactory mapperFactory, Authenticator authenticator, MailboxPathLocker locker) {
+ super(mapperFactory, authenticator, locker);
+ }
+
+ public HBaseMailboxManager(HBaseMailboxSessionMapperFactory mapperFactory, Authenticator authenticator) {
+ super(mapperFactory, authenticator, new JVMMailboxPathLocker());
+ }
+
+ @Override
+ protected Mailbox<UUID> doCreateMailbox(MailboxPath mailboxPath, MailboxSession session) throws MailboxException {
+ return new HBaseMailbox(mailboxPath, randomUidValidity());
+ }
+
+ /**
+ * Delete all mailboxes
+ *
+ * @param mailboxSession
+ * @throws MailboxException
+ */
+ public void deleteEverything(MailboxSession mailboxSession) throws MailboxException {
+
+ final HBaseMailboxMapper mapper = (HBaseMailboxMapper) getMapperFactory().getMailboxMapper(mailboxSession);
+
+ mapper.execute(new TransactionalMapper.VoidTransaction() {
+
+ @Override
+ public void runVoid() throws MailboxException {
+ mapper.deleteAllMemberships();
+ }
+ });
+ mapper.execute(new TransactionalMapper.VoidTransaction() {
+
+ @Override
+ public void runVoid() throws MailboxException {
+ mapper.deleteAllMailboxes();
+ }
+ });
+ }
+
+ @Override
+ protected StoreMessageManager<UUID> createMessageManager(Mailbox<UUID> mailboxRow, MailboxSession session) throws MailboxException {
+ StoreMessageManager<UUID> result = new HBaseMessageManager(getMapperFactory(), getMessageSearchIndex(), getEventDispatcher(), mailboxRow);
+ return result;
+ }
+}
Added: james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/HBaseMailboxSessionMapperFactory.java
URL: http://svn.apache.org/viewvc/james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/HBaseMailboxSessionMapperFactory.java?rev=1164966&view=auto
==============================================================================
--- james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/HBaseMailboxSessionMapperFactory.java (added)
+++ james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/HBaseMailboxSessionMapperFactory.java Sun Sep 4 07:31:41 2011
@@ -0,0 +1,155 @@
+/****************************************************************
+ * 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.james.mailbox.hbase;
+
+import java.io.IOException;
+import java.util.UUID;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HColumnDescriptor;
+import org.apache.hadoop.hbase.HTableDescriptor;
+import org.apache.hadoop.hbase.MasterNotRunningException;
+import org.apache.hadoop.hbase.ZooKeeperConnectionException;
+import org.apache.hadoop.hbase.client.HBaseAdmin;
+import org.apache.james.mailbox.MailboxException;
+import org.apache.james.mailbox.MailboxSession;
+import org.apache.james.mailbox.SubscriptionException;
+import org.apache.james.mailbox.hbase.mail.HBaseMailboxMapper;
+import org.apache.james.mailbox.hbase.mail.HBaseMessageMapper;
+import org.apache.james.mailbox.hbase.user.HBaseSubscriptionMapper;
+import org.apache.james.mailbox.store.MailboxSessionMapperFactory;
+import org.apache.james.mailbox.store.mail.MailboxMapper;
+import org.apache.james.mailbox.store.mail.MessageMapper;
+import org.apache.james.mailbox.store.mail.ModSeqProvider;
+import org.apache.james.mailbox.store.mail.UidProvider;
+import org.apache.james.mailbox.store.user.SubscriptionMapper;
+
+import static org.apache.james.mailbox.hbase.HBaseNames.*;
+
+/**
+ * HBase implementation of {@link MailboxSessionMapperFactory}
+ *
+ */
+public class HBaseMailboxSessionMapperFactory extends MailboxSessionMapperFactory<UUID> {
+
+ private final Configuration conf;
+ private final UidProvider<UUID> uidProvider;
+ private final ModSeqProvider<UUID> modSeqProvider;
+
+ /**
+ * Creates the necessary tables in HBase if they do not exist.
+ *
+ * @param conf Configuration for the cluster
+ * @param uidProvider UID provider for mailbox uid.
+ * @param modSeqProvider
+ * @throws MasterNotRunningException
+ * @throws ZooKeeperConnectionException
+ * @throws IOException
+ */
+ public HBaseMailboxSessionMapperFactory(Configuration conf, UidProvider<UUID> uidProvider, ModSeqProvider<UUID> modSeqProvider) {
+ this.conf = conf;
+ this.uidProvider = uidProvider;
+ this.modSeqProvider = modSeqProvider;
+
+ //TODO: add better exception handling for this
+ try {
+ HBaseAdmin hbaseAdmin = new HBaseAdmin(conf);
+ HTableDescriptor desc = null;
+ HColumnDescriptor hColumnDescriptor = null;
+
+ /* create the tables if it does not exist */
+
+ if (!hbaseAdmin.tableExists(MAILBOXES_TABLE)) {
+ desc = new HTableDescriptor(MAILBOXES_TABLE);
+ hColumnDescriptor = new HColumnDescriptor(MAILBOX_CF);
+ hColumnDescriptor.setMaxVersions(1);
+ desc.addFamily(hColumnDescriptor);
+ hbaseAdmin.createTable(desc);
+ }
+
+ if (!hbaseAdmin.tableExists(MESSAGES_TABLE)) {
+ /**TODO: try to reduce the number of column families as suggested by:
+ * http://hbase.apache.org/book.html#number.of.cfs
+ * Down to three column families, striking for just two.
+ */
+ desc = new HTableDescriptor(MESSAGES_TABLE);
+ hColumnDescriptor = new HColumnDescriptor(MESSAGES_META_CF);
+ hColumnDescriptor.setMaxVersions(1);
+ desc.addFamily(hColumnDescriptor);
+ hColumnDescriptor = new HColumnDescriptor(MESSAGE_DATA_HEADERS);
+ hColumnDescriptor.setMaxVersions(1);
+ desc.addFamily(hColumnDescriptor);
+ hColumnDescriptor = new HColumnDescriptor(MESSAGE_DATA_BODY);
+ hColumnDescriptor.setMaxVersions(1);
+ desc.addFamily(hColumnDescriptor);
+ hbaseAdmin.createTable(desc);
+ }
+
+
+ if (!hbaseAdmin.tableExists(SUBSCRIPTIONS_TABLE)) {
+ desc = new HTableDescriptor(SUBSCRIPTIONS_TABLE);
+ hColumnDescriptor = new HColumnDescriptor(SUBSCRIPTION_CF);
+ hColumnDescriptor.setMaxVersions(1);
+ desc.addFamily(hColumnDescriptor);
+ hbaseAdmin.createTable(desc);
+ }
+
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ protected MessageMapper<UUID> createMessageMapper(MailboxSession session) throws MailboxException {
+ return new HBaseMessageMapper(session, uidProvider, modSeqProvider, this.conf);
+ }
+
+ @Override
+ protected MailboxMapper<UUID> createMailboxMapper(MailboxSession session) throws MailboxException {
+ return new HBaseMailboxMapper(this.conf);
+ }
+
+ @Override
+ protected SubscriptionMapper createSubscriptionMapper(MailboxSession session) throws SubscriptionException {
+ return new HBaseSubscriptionMapper(this.conf);
+ }
+
+ /**
+ * Returns the configuration object for accesing the cluster.
+ * @return
+ */
+ public Configuration getClusterConfiguration() {
+ return conf;
+ }
+
+ /**
+ * Returns the ModSeqProvider used.
+ * @return
+ */
+ public ModSeqProvider<UUID> getModSeqProvider() {
+ return modSeqProvider;
+ }
+
+ /**
+ * Returns the UidProvider that generates UID's for mailboxes.
+ * @return
+ */
+ public UidProvider<UUID> getUidProvider() {
+ return uidProvider;
+ }
+}
Added: james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/HBaseMessageManager.java
URL: http://svn.apache.org/viewvc/james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/HBaseMessageManager.java?rev=1164966&view=auto
==============================================================================
--- james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/HBaseMessageManager.java (added)
+++ james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/HBaseMessageManager.java Sun Sep 4 07:31:41 2011
@@ -0,0 +1,52 @@
+/****************************************************************
+ * 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.james.mailbox.hbase;
+
+import java.util.UUID;
+import javax.mail.Flags;
+import org.apache.james.mailbox.MailboxException;
+import org.apache.james.mailbox.MailboxSession;
+import org.apache.james.mailbox.store.MailboxEventDispatcher;
+import org.apache.james.mailbox.store.MailboxSessionMapperFactory;
+import org.apache.james.mailbox.store.StoreMessageManager;
+import org.apache.james.mailbox.store.mail.model.Mailbox;
+import org.apache.james.mailbox.store.search.MessageSearchIndex;
+
+/**
+ * HBase implementation of MessageManager.
+ *
+ */
+public class HBaseMessageManager extends StoreMessageManager<UUID> {
+
+ public HBaseMessageManager(MailboxSessionMapperFactory<UUID> mapperFactory, MessageSearchIndex<UUID> index,
+ MailboxEventDispatcher<UUID> dispatcher, Mailbox<UUID> mailbox) throws MailboxException {
+ super(mapperFactory, index, dispatcher, mailbox);
+
+ }
+
+ /**
+ * Support user flags
+ */
+ @Override
+ protected Flags getPermanentFlags(MailboxSession session) {
+ Flags flags = super.getPermanentFlags(session);
+ flags.add(Flags.Flag.USER);
+ return flags;
+ }
+}
Added: james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/HBaseNames.java
URL: http://svn.apache.org/viewvc/james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/HBaseNames.java?rev=1164966&view=auto
==============================================================================
--- james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/HBaseNames.java (added)
+++ james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/HBaseNames.java Sun Sep 4 07:31:41 2011
@@ -0,0 +1,72 @@
+/*
+ * To change this template, choose Tools | Templates
+ * and open the template in the editor.
+ */
+package org.apache.james.mailbox.hbase;
+
+import org.apache.hadoop.hbase.util.Bytes;
+
+/**
+ * Contains table names, column family names, qualifier names and other constants
+ * for use in HBase.
+ *
+ * Each qualifier in the META column will begin with a short prefix that will
+ * determine it's purpose. </br>
+ * Qualifier prefix meaning:<br/>
+ * <ul>
+ * <li> m: meta information; </li>
+ * <li> sf: system flag (DELETE, RECENT, etc.) </li>
+ * <li> uf: user flag </li>
+ * <li> p: user property</li>
+ * </ul>
+ */
+public interface HBaseNames {
+
+ /**
+ * The HBase table name for storing mailbox names
+ */
+ public static final String MAILBOXES = "JAMES_MAILBOXES";
+ public static final byte[] MAILBOXES_TABLE = Bytes.toBytes(MAILBOXES);
+ /** Default mailbox column family */
+ public static final byte[] MAILBOX_CF = Bytes.toBytes("DATA");
+ /** HBase column qualifiers: field names stored as byte arrays*/
+ public static final byte[] MAILBOX_NAME = Bytes.toBytes("name");
+ public static final byte[] MAILBOX_USER = Bytes.toBytes("user");
+ public static final byte[] MAILBOX_NAMESPACE = Bytes.toBytes("namespace");
+ public static final byte[] MAILBOX_LASTUID = Bytes.toBytes("lastUID");
+ public static final byte[] MAILBOX_UIDVALIDITY = Bytes.toBytes("uidValidity");
+ public static final byte[] MAILBOX_HIGHEST_MODSEQ = Bytes.toBytes("hModSeq");
+ public static final byte[] MAILBOX_MESSAGE_COUNT = Bytes.toBytes("count");
+ /** The HBase table name for storing subscriptions */
+ public static final String SUBSCRIPTIONS = "JAMES_SUBSCRIPTIONS";
+ /** The HBase table name for storing subscriptions */
+ public static final byte[] SUBSCRIPTIONS_TABLE = Bytes.toBytes(SUBSCRIPTIONS);
+ /** Default subscription column family */
+ public static final byte[] SUBSCRIPTION_CF = Bytes.toBytes("DATA");
+ /** The HBase table name for storing messages */
+ public static final String MESSAGES = "JAMES_MESSAGES";
+ /** The HBase table name for storing messages */
+ public static final byte[] MESSAGES_TABLE = Bytes.toBytes(MESSAGES);
+ /** Column family for storing message data*/
+ //public static final byte[] MESSAGES_DATA_CF = Bytes.toBytes("DATA");
+ /** Column family for storing message meta information*/
+ public static final byte[] MESSAGES_META_CF = Bytes.toBytes("META");
+ /** Column family for storing message headers*/
+ public static final byte[] MESSAGE_DATA_HEADERS = Bytes.toBytes("HEAD");
+ /** Column family for storing message body*/
+ public static final byte[] MESSAGE_DATA_BODY = Bytes.toBytes("BODY");
+ public static final String PREFIX_META = "m:";
+ public static final byte[] PREFIX_META_B = Bytes.toBytes(PREFIX_META);
+ /** kept sorted */
+ public static final byte[] MESSAGE_BODY_OCTETS = Bytes.toBytes(PREFIX_META + "body");
+ public static final byte[] MESSAGE_CONTENT_OCTETS = Bytes.toBytes(PREFIX_META + "content");
+ public static final byte[] MESSAGE_INTERNALDATE = Bytes.toBytes(PREFIX_META + "date");
+ public static final byte[] MESSAGE_TEXT_LINE_COUNT = Bytes.toBytes(PREFIX_META + "lcount");
+ public static final byte[] MESSAGE_MODSEQ = Bytes.toBytes(PREFIX_META + "mseq");
+ public static final byte[] MESSAGE_MEDIA_TYPE = Bytes.toBytes(PREFIX_META + "mtype");
+ public static final byte[] MESSAGE_SUB_TYPE = Bytes.toBytes(PREFIX_META + "stype");
+ public static final byte[] MARKER_PRESENT = Bytes.toBytes("X");
+ public static final byte[] MARKER_MISSING = Bytes.toBytes(" ");
+ // the maximum recomended HBase column size is 10 MB
+ public static final int MAX_COLUMN_SIZE = 1024; //2 * 1024 * 1024;
+}
Added: james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/HBaseNonTransactionalMapper.java
URL: http://svn.apache.org/viewvc/james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/HBaseNonTransactionalMapper.java?rev=1164966&view=auto
==============================================================================
--- james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/HBaseNonTransactionalMapper.java (added)
+++ james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/HBaseNonTransactionalMapper.java Sun Sep 4 07:31:41 2011
@@ -0,0 +1,39 @@
+/****************************************************************
+ * 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.james.mailbox.hbase;
+
+import org.apache.james.mailbox.store.transaction.NonTransactionalMapper;
+
+/**
+ * HBase implementation of TransactionMapper.
+ * I don't know if this class is thread-safe!
+ * Assume it is not!
+ *
+ */
+public class HBaseNonTransactionalMapper extends NonTransactionalMapper {
+
+ /**
+ * End request
+ */
+ @Override
+ public void endRequest() {
+ //TODO: maybe do some thing more wise here?
+ //System.out.println("Bye!");
+ }
+}
Added: james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/HBaseUtils.java
URL: http://svn.apache.org/viewvc/james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/HBaseUtils.java?rev=1164966&view=auto
==============================================================================
--- james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/HBaseUtils.java (added)
+++ james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/HBaseUtils.java Sun Sep 4 07:31:41 2011
@@ -0,0 +1,436 @@
+/****************************************************************
+ * 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.james.mailbox.hbase;
+
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+import java.util.NavigableMap;
+import java.util.UUID;
+import javax.mail.Flags;
+import javax.mail.Flags.Flag;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.client.Delete;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.james.mailbox.MailboxPath;
+import org.apache.james.mailbox.hbase.io.ChunkInputStream;
+import org.apache.james.mailbox.hbase.mail.model.HBaseMailbox;
+import org.apache.james.mailbox.hbase.mail.HBaseMessage;
+import org.apache.james.mailbox.store.mail.model.Mailbox;
+import org.apache.james.mailbox.store.mail.model.Message;
+import org.apache.james.mailbox.store.mail.model.Property;
+import org.apache.james.mailbox.store.mail.model.impl.PropertyBuilder;
+import org.apache.james.mailbox.store.user.model.Subscription;
+
+import static org.apache.james.mailbox.hbase.FlagConvertor.*;
+import static org.apache.james.mailbox.hbase.PropertyConvertor.*;
+import static org.apache.james.mailbox.hbase.HBaseNames.*;
+
+/**
+ * HBase utility classes for mailbox and message manipulation.
+ * @author ieugen
+ */
+public class HBaseUtils {
+ // TODO: swith to a bit wise implementation of flags.
+
+ /**
+ * Creates a Mailbox object from a HBase Result object.
+ * @param result a result of a HBase Get operation
+ * @return a Mailbox object
+ */
+ public static Mailbox<UUID> mailboxFromResult(Result result) {
+ NavigableMap<byte[], byte[]> rawMailbox = result.getFamilyMap(MAILBOX_CF);
+ //TODO: should we test for null values?
+ MailboxPath path = new MailboxPath(Bytes.toString(rawMailbox.get(MAILBOX_NAMESPACE)),
+ Bytes.toString(rawMailbox.get(MAILBOX_USER)),
+ Bytes.toString(rawMailbox.get(MAILBOX_NAME)));
+
+ HBaseMailbox mailbox = new HBaseMailbox(path, Bytes.toLong(rawMailbox.get(MAILBOX_UIDVALIDITY)));
+ mailbox.setMailboxId(UUIDFromRowKey(result.getRow()));
+ mailbox.setHighestModSeq(Bytes.toLong(rawMailbox.get(MAILBOX_HIGHEST_MODSEQ)));
+ mailbox.setLastUid(Bytes.toLong(rawMailbox.get(MAILBOX_LASTUID)));
+ mailbox.setMessageCount(Bytes.toLong(rawMailbox.get(MAILBOX_MESSAGE_COUNT)));
+ return mailbox;
+ }
+
+ /**
+ * This returns the row key needed for HBase. Having the method here ensure
+ * we have a consistent way to generate the rowkey.
+ * @return rowkey byte array that can be used with HBase API
+ */
+ /**
+ * Convenience method for generating a rowKey when you don't have a mailbox object.
+ * @param uuid
+ * @return
+ */
+ public static byte[] mailboxRowKey(UUID uuid) {
+ byte[] rowKey = new byte[16];
+ int offset = Bytes.putLong(rowKey, 0, uuid.getMostSignificantBits());
+ Bytes.putLong(rowKey, offset, uuid.getLeastSignificantBits());
+ return rowKey;
+ }
+
+ /**
+ * Returns a UUID from the a byte array.
+ * @param rowkey
+ * @return
+ */
+ public static UUID UUIDFromRowKey(byte[] rowkey) {
+ return new UUID(Bytes.toLong(rowkey, 0), Bytes.toLong(rowkey, 8));
+ }
+
+ /**
+ * Transforms the mailbox into a Put operation.
+ * @return a Put object
+ */
+ public static Put toPut(HBaseMailbox mailbox) {
+ Put put = new Put(mailboxRowKey(mailbox.getMailboxId()));
+ // we don't store null values and we don't restore them. it's a column based store.
+ if (mailbox.getName() != null) {
+ put.add(MAILBOX_CF, MAILBOX_NAME, Bytes.toBytes(mailbox.getName()));
+ }
+
+ if (mailbox.getUser() != null) {
+ put.add(MAILBOX_CF, MAILBOX_USER, Bytes.toBytes(mailbox.getUser()));
+ }
+ if (mailbox.getNamespace() != null) {
+ put.add(MAILBOX_CF, MAILBOX_NAMESPACE, Bytes.toBytes(mailbox.getNamespace()));
+ }
+ put.add(MAILBOX_CF, MAILBOX_LASTUID, Bytes.toBytes(mailbox.getLastUid()));
+ put.add(MAILBOX_CF, MAILBOX_UIDVALIDITY, Bytes.toBytes(mailbox.getUidValidity()));
+ put.add(MAILBOX_CF, MAILBOX_HIGHEST_MODSEQ, Bytes.toBytes(mailbox.getHighestModSeq()));
+ put.add(MAILBOX_CF, MAILBOX_MESSAGE_COUNT, Bytes.toBytes(mailbox.getMessageCount()));
+ return put;
+ }
+
+ /**
+ * Transforms only the metadata into a Put object. The rest of the message will
+ * be transfered using multiple Puts if size requires it.
+ * @param message
+ * @return a put that contains all metadata information.
+ */
+ public static Put metadataToPut(Message<UUID> message) {
+ Put put = new Put(messageRowKey(message));
+ // we store the message uid and mailbox uid in the row key
+ // store the meta data
+ put.add(MESSAGES_META_CF, MESSAGE_MODSEQ, Bytes.toBytes(message.getModSeq()));
+ put.add(MESSAGES_META_CF, MESSAGE_INTERNALDATE, Bytes.toBytes(message.getInternalDate().getTime()));
+ put.add(MESSAGES_META_CF, MESSAGE_MEDIA_TYPE, Bytes.toBytes(message.getMediaType()));
+ put.add(MESSAGES_META_CF, MESSAGE_SUB_TYPE, Bytes.toBytes(message.getSubType()));
+ put.add(MESSAGES_META_CF, MESSAGE_CONTENT_OCTETS, Bytes.toBytes(message.getFullContentOctets()));
+ put.add(MESSAGES_META_CF, MESSAGE_BODY_OCTETS, Bytes.toBytes(message.getBodyOctets()));
+ if (message.getTextualLineCount() != null) {
+ put.add(MESSAGES_META_CF, MESSAGE_TEXT_LINE_COUNT, Bytes.toBytes(message.getTextualLineCount()));
+ }
+ // store system flags in meta and user flags in uflags to avoid name clashes
+ Flags flags = message.createFlags();
+ // system flags
+ if (flags.contains(Flag.ANSWERED)) {
+ put.add(MESSAGES_META_CF, FLAGS_ANSWERED, MARKER_PRESENT);
+ }
+ if (flags.contains(Flag.DELETED)) {
+ put.add(MESSAGES_META_CF, FLAGS_DELETED, MARKER_PRESENT);
+ }
+ if (flags.contains(Flag.DRAFT)) {
+ put.add(MESSAGES_META_CF, FLAGS_DRAFT, MARKER_PRESENT);
+ }
+ if (flags.contains(Flag.FLAGGED)) {
+ put.add(MESSAGES_META_CF, FLAGS_FLAGGED, MARKER_PRESENT);
+ }
+ if (flags.contains(Flag.RECENT)) {
+ put.add(MESSAGES_META_CF, FLAGS_RECENT, MARKER_PRESENT);
+ }
+ if (flags.contains(Flag.SEEN)) {
+ put.add(MESSAGES_META_CF, FLAGS_SEEN, MARKER_PRESENT);
+ }
+ if (flags.contains(Flag.USER)) {
+ put.add(MESSAGES_META_CF, FLAGS_USER, MARKER_PRESENT);
+ }
+
+ // user flags
+ for (String flag : flags.getUserFlags()) {
+ put.add(MESSAGES_META_CF, userFlagToBytes(flag), MARKER_PRESENT);
+ }
+ int propNumber = 0;
+ // add the properties
+ for (Property prop : message.getProperties()) {
+ put.add(MESSAGES_META_CF, getQualifier(propNumber++), getValue(prop));
+ }
+
+ return put;
+ }
+
+ /**
+ * Create a row key for a message in a mailbox. The current row key is mailboxID followed by messageID.
+ * Both values are fixed length so no separator is needed.
+ * Downside: we will be storing the same message multiple times, one time for each recipient.
+ * @param message
+ * @return a row key
+ */
+ public static byte[] messageRowKey(Message<UUID> message) {
+ return messageRowKey(message.getMailboxId(), message.getUid());
+ }
+
+ /**
+ * Utility method to build row keys from mailbox UUID and message uid.
+ * The message uid's are stored in reverse order by substracting the uid value
+ * from Long.MAX_VALUE.
+ * @param message
+ * @param uid
+ * @return a row key byte array
+ */
+ public static byte[] messageRowKey(UUID mailboxUid, long uid) {
+ /** message uid's are stored in reverse order so we will always have the most recent messages first*/
+ byte[] ba = Bytes.add(Bytes.toBytes(mailboxUid.getMostSignificantBits()),
+ Bytes.toBytes(mailboxUid.getLeastSignificantBits()),
+ Bytes.toBytes(Long.MAX_VALUE - uid));
+ //System.out.println(Bytes.toStringBinary(ba));
+ return ba;
+ }
+
+ /**
+ * Utility to build row keys from mailboxUID and a value. The value is added to
+ * the key without any other opperations.
+ * @param mailboxUid
+ * @param value
+ * @return
+ */
+ public static byte[] customMessageRowKey(UUID mailboxUid, long value) {
+ return Bytes.add(Bytes.toBytes(mailboxUid.getMostSignificantBits()),
+ Bytes.toBytes(mailboxUid.getLeastSignificantBits()),
+ Bytes.toBytes(value));
+ }
+
+ /**
+ * Creates a HBaseMessage from a Result object. This method retrieves all information
+ * except for body and header related bytes. The message content will be loaded on demand
+ * through a specialised InputStream called {@link ChunkInputStream}.
+ * IMPORTANT: the method expects a single version of each cell. Use setMaxVersions(1).
+ * @param conf configuration object for HBase cluster
+ * @param result the result object containing message data
+ * @return a HBaseMessage instance with message metadata.
+ */
+ public static Message<UUID> messageMetaFromResult(Configuration conf, Result result) {
+ HBaseMessage message = null;
+ Flags flags = new Flags();
+ List<Property> propList = new ArrayList<Property>();
+ KeyValue[] keys = result.raw();
+ String mediaType = null, subType = null;
+ Long modSeq = null, uid, bodyOctets = null, contentOctets = null, textualLineCount = null;
+ Date internalDate = null;
+
+ int i = 0;
+ /** it is VERY IMPORTANT that the byte arrays are kept ascending */
+ if (Bytes.equals(keys[i].getQualifier(), MESSAGE_BODY_OCTETS)) {
+ bodyOctets = Bytes.toLong(keys[i].getValue());
+ i++;
+ }
+ if (Bytes.equals(keys[i].getQualifier(), MESSAGE_CONTENT_OCTETS)) {
+ contentOctets = Bytes.toLong(keys[i].getValue());
+ i++;
+ }
+ if (Bytes.equals(keys[i].getQualifier(), MESSAGE_INTERNALDATE)) {
+ internalDate = new Date(Bytes.toLong(keys[i].getValue()));
+ i++;
+ }
+ // may be null so it will probably skip
+ if (Bytes.equals(keys[i].getQualifier(), MESSAGE_TEXT_LINE_COUNT)) {
+ textualLineCount = Bytes.toLong(keys[i].getValue());
+ i++;
+ }
+
+ if (Bytes.equals(keys[i].getQualifier(), MESSAGE_MODSEQ)) {
+ modSeq = Bytes.toLong(keys[i].getValue());
+ i++;
+ }
+ if (Bytes.equals(keys[i].getQualifier(), MESSAGE_MEDIA_TYPE)) {
+ mediaType = Bytes.toString(keys[i].getValue());
+ i++;
+ }
+ if (Bytes.equals(keys[i].getQualifier(), MESSAGE_SUB_TYPE)) {
+ subType = Bytes.toString(keys[i].getValue());
+ i++;
+ }
+ // only TEXT_LINE_COUNT can be missing if message is binary
+ if (i < 5) {
+ throw new RuntimeException("HBase message column names not sorted.");
+ }
+ while (i < keys.length) {
+ //get message properties
+ if (Bytes.startsWith(keys[i].getQualifier(), PREFIX_PROP_B)) {
+ propList.add(getProperty(keys[i].getValue()));
+ } else if (Bytes.startsWith(keys[i].getQualifier(), PREFIX_SFLAGS_B)) {
+ // get system flags, stored as qualifiers
+ if (Bytes.equals(MARKER_PRESENT, keys[i].getValue())) {
+ flags.add(systemFlagFromBytes(keys[i].getQualifier()));
+ }
+ } else if (Bytes.startsWith(keys[i].getQualifier(), PREFIX_UFLAGS_B)) {
+ // get user flags, stored as value qualifier
+ flags.add(userFlagFromBytes(keys[i].getQualifier()));
+ }
+ i++;
+ }
+ UUID uuid = UUIDFromRowKey(result.getRow());
+ uid = Long.MAX_VALUE - Bytes.toLong(result.getRow(), 16);
+ PropertyBuilder props = new PropertyBuilder(propList);
+ props.setMediaType(mediaType);
+ props.setSubType(subType);
+ message = new HBaseMessage(conf, uuid, internalDate, flags, contentOctets, (int) (contentOctets - bodyOctets), props);
+ message.setUid(uid);
+ message.setModSeq(modSeq);
+ message.setTextualLineCount(textualLineCount);
+ return message;
+ }
+
+ /**
+ * Creates a Put object from this subscription object
+ * @return Put object suitable for HBase persistence
+ */
+ public static Put toPut(Subscription subscription) {
+ Put put = new Put(Bytes.toBytes(subscription.getUser()));
+ put.add(SUBSCRIPTION_CF, Bytes.toBytes(subscription.getMailbox()), MARKER_PRESENT);
+ return put;
+ }
+
+ /**
+ * Utility method to transform message flags into a put opperation.
+ * @param message
+ * @param flags
+ * @return a put object with
+ */
+ public static Put flagsToPut(Message<UUID> message, Flags flags) {
+ Put put = new Put(messageRowKey(message));
+ //system flags
+ if (flags.contains(Flag.ANSWERED)) {
+ put.add(MESSAGES_META_CF, FLAGS_ANSWERED, MARKER_PRESENT);
+ } else {
+ put.add(MESSAGES_META_CF, FLAGS_ANSWERED, MARKER_MISSING);
+ }
+ if (flags.contains(Flag.DELETED)) {
+ put.add(MESSAGES_META_CF, FLAGS_DELETED, MARKER_PRESENT);
+ } else {
+ put.add(MESSAGES_META_CF, FLAGS_DELETED, MARKER_MISSING);
+ }
+ if (flags.contains(Flag.DRAFT)) {
+ put.add(MESSAGES_META_CF, FLAGS_DRAFT, MARKER_PRESENT);
+ } else {
+ put.add(MESSAGES_META_CF, FLAGS_DRAFT, MARKER_MISSING);
+ }
+ if (flags.contains(Flag.FLAGGED)) {
+ put.add(MESSAGES_META_CF, FLAGS_FLAGGED, MARKER_PRESENT);
+ } else {
+ put.add(MESSAGES_META_CF, FLAGS_FLAGGED, MARKER_MISSING);
+ }
+ if (flags.contains(Flag.RECENT)) {
+ put.add(MESSAGES_META_CF, FLAGS_RECENT, MARKER_PRESENT);
+ } else {
+ put.add(MESSAGES_META_CF, FLAGS_RECENT, MARKER_MISSING);
+ }
+ if (flags.contains(Flag.SEEN)) {
+ put.add(MESSAGES_META_CF, FLAGS_SEEN, MARKER_PRESENT);
+ } else {
+ put.add(MESSAGES_META_CF, FLAGS_SEEN, MARKER_MISSING);
+ }
+ if (flags.contains(Flag.USER)) {
+ put.add(MESSAGES_META_CF, FLAGS_USER, MARKER_PRESENT);
+ } else {
+ put.add(MESSAGES_META_CF, FLAGS_USER, MARKER_MISSING);
+ }
+ /**TODO: user flags are not deleted this way: store them all in a single column
+ * and replace that column full.
+ */
+ // user flags
+ for (String flag : flags.getUserFlags()) {
+ put.add(MESSAGES_META_CF, userFlagToBytes(flag), MARKER_PRESENT);
+ }
+ return put;
+ }
+
+ public static Delete flagsToDelete(Message<UUID> message, Flags flags) {
+ Delete delete = new Delete(messageRowKey(message));
+ //we mark for delete flags that are not present (they will be Put'ed)
+ if (flags.contains(Flag.ANSWERED)) {
+ delete.deleteColumn(MESSAGES_META_CF, FLAGS_ANSWERED);
+ }
+ if (flags.contains(Flag.DELETED)) {
+ delete.deleteColumn(MESSAGES_META_CF, FLAGS_DELETED);
+ }
+ if (flags.contains(Flag.DRAFT)) {
+ delete.deleteColumn(MESSAGES_META_CF, FLAGS_DRAFT);
+ }
+ if (flags.contains(Flag.FLAGGED)) {
+ delete.deleteColumn(MESSAGES_META_CF, FLAGS_FLAGGED);
+ }
+ if (flags.contains(Flag.RECENT)) {
+ delete.deleteColumn(MESSAGES_META_CF, FLAGS_RECENT);
+ }
+ if (flags.contains(Flag.SEEN)) {
+ delete.deleteColumn(MESSAGES_META_CF, FLAGS_SEEN);
+ }
+ if (flags.contains(Flag.USER)) {
+ delete.deleteColumn(MESSAGES_META_CF, FLAGS_USER);
+ }
+
+ // we delete all user flags that where not in the new configuration
+ for (String flag : flags.getUserFlags()) {
+ delete.deleteColumn(MESSAGES_META_CF, userFlagToBytes(flag));
+ }
+ return delete;
+ }
+
+ /**
+ * Returns a String composed of all flags in the parameter.
+ * @param flags
+ * @return a string representation of all flags
+ */
+ public static String flagsToString(Flags flags) {
+ StringBuilder b = new StringBuilder();
+
+ if (flags.contains(Flag.ANSWERED)) {
+ b.append("ANSWERED ");
+ }
+ if (flags.contains(Flag.DELETED)) {
+ b.append("DELETED ");
+ }
+ if (flags.contains(Flag.DRAFT)) {
+ b.append("DRAFT ");
+ }
+ if (flags.contains(Flag.FLAGGED)) {
+ b.append("FLAGGED ");
+ }
+ if (flags.contains(Flag.RECENT)) {
+ b.append("RECENT ");
+ }
+ if (flags.contains(Flag.SEEN)) {
+ b.append("SEEN ");
+ }
+ if (flags.contains(Flag.USER)) {
+ b.append("USER ");
+ }
+ for (String flag : flags.getUserFlags()) {
+ b.append(flag);
+ b.append(" ");
+ }
+ return b.toString();
+ }
+}
Added: james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/PropertyConvertor.java
URL: http://svn.apache.org/viewvc/james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/PropertyConvertor.java?rev=1164966&view=auto
==============================================================================
--- james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/PropertyConvertor.java (added)
+++ james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/PropertyConvertor.java Sun Sep 4 07:31:41 2011
@@ -0,0 +1,74 @@
+/****************************************************************
+ * 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.james.mailbox.hbase;
+
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.james.mailbox.store.mail.model.Property;
+import org.apache.james.mailbox.store.mail.model.impl.SimpleProperty;
+
+/**
+ * Class used for converting message properties to and from byte arrays for use as
+ * HBase column qualifiers and values.
+ */
+public class PropertyConvertor {
+
+ public static final String PREFIX_PROP = "p:";
+ public static final byte[] PREFIX_PROP_B = Bytes.toBytes(PREFIX_PROP);
+ //TODO: find a better separator.
+ /** The separator must not be part of the property namespace or localName */
+ private static final String SEPARATOR = "%%";
+
+ /**
+ * Returns a byte array that represents a HBase column qualifier for the
+ * provided property.
+ * @param prop the property for storage n HBase
+ * @return a byte array that represents a column qualifier for the property
+ */
+ public static byte[] getQualifier(int propNumber) {
+ // allow for about 1000 properties to be stored, we pad them because HBase will store them sorted
+ return Bytes.toBytes(PREFIX_PROP + String.format("%03d", propNumber));
+ }
+
+ /**
+ * Returns a byte array representation of the Property value.
+ * (uses Bytes.toBytes)
+ * @param prop
+ * @return a byte array of the value.
+ */
+ public static byte[] getValue(Property prop) {
+ return Bytes.toBytes(prop.getNamespace() + SEPARATOR + prop.getLocalName() + SEPARATOR + prop.getValue());
+ }
+
+ /**
+ * Returns a Property from a qualifier byte array.
+ * @param value
+ * @return a {@link Property}
+ * @throws RuntimeException if property prefix or separator is not present
+ */
+ public static Property getProperty(byte[] value) {
+ String ns = Bytes.toString(value);
+ //TODO: we assume the SEPARATOR=%% can not appear in a normal property. This may not be true.
+ String[] parts = ns.split(SEPARATOR);
+ if (parts.length != 3) {
+ throw new RuntimeException("Separator not found in qualifier "
+ + Bytes.toString(value));
+ }
+ return new SimpleProperty(parts[0], parts[1], parts[2]);
+ }
+}
Added: james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/io/ChunkInputStream.java
URL: http://svn.apache.org/viewvc/james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/io/ChunkInputStream.java?rev=1164966&view=auto
==============================================================================
--- james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/io/ChunkInputStream.java (added)
+++ james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/io/ChunkInputStream.java Sun Sep 4 07:31:41 2011
@@ -0,0 +1,114 @@
+/****************************************************************
+ * 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.james.mailbox.hbase.io;
+
+import java.io.IOException;
+import java.io.InputStream;
+import org.apache.hadoop.hbase.client.HTable;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.client.Get;
+import org.apache.hadoop.hbase.client.Result;
+
+/**
+ * Return an InputStream which retrieve columns from a row which stores chunk of
+ * data. See also {@link ChunkOutputStream}
+ *
+ * This implementation is not thread-safe!
+ *
+ * Bsed on Hector implementation for Cassandra.
+ * https://github.com/rantav/hector/blob/master/core/src/main/java/me/prettyprint/cassandra/io/ChunkInputStream.java
+ */
+public class ChunkInputStream extends InputStream {
+
+ private final Configuration conf;
+ private final byte[] tableName;
+ private final byte[] cf;
+ private final byte[] key;
+ private int pos;
+ private long chunkPos = 1;
+ private byte[] chunk;
+
+ public ChunkInputStream(Configuration conf, byte[] tableName, byte[] cf, byte[] key) {
+ this.key = key;
+ this.conf = conf;
+ this.tableName = tableName;
+ this.cf = cf;
+ }
+
+ public ChunkInputStream(Configuration conf, String tableName, String cf, byte[] key) {
+ this(conf, Bytes.toBytes(tableName), Bytes.toBytes(cf), key);
+ }
+
+ /*
+ * (non-Javadoc)
+ *
+ * @see java.io.InputStream#read()
+ */
+ @Override
+ public int read() throws IOException {
+ if (chunk == null || pos + 1 == chunk.length) {
+ if (!fetchChunk()) {
+ return -1;
+ }
+ }
+ return chunk[pos++];
+ }
+
+ /**
+ * Fetch the next chunk.
+ *
+ * @return exists if there was a chunk to fetch.
+ * @throws IOException
+ */
+ private boolean fetchChunk() throws IOException {
+ HTable messages = null;
+ try {
+ byte[] cp = Bytes.toBytes(chunkPos);
+ messages = new HTable(conf, tableName);
+ Get get = new Get(key);
+ get.addColumn(cf, cp);
+ get.setMaxVersions(1);
+ Result result = messages.get(get);
+ if (!result.isEmpty()) {
+ chunk = result.getValue(cf, cp);
+ chunkPos++;
+ pos = 0;
+ return true;
+ } else {
+ return false;
+ }
+ } catch (IOException e) {
+ throw new IOException("Unable to read data", e);
+ } finally {
+ if (messages != null) {
+ messages.close();
+
+ }
+ }
+ }
+
+ /**
+ * Not supported
+ */
+ @Override
+ public boolean markSupported() {
+ return false;
+ }
+}
Added: james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/io/ChunkOutputStream.java
URL: http://svn.apache.org/viewvc/james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/io/ChunkOutputStream.java?rev=1164966&view=auto
==============================================================================
--- james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/io/ChunkOutputStream.java (added)
+++ james/mailbox/tags/apache-james-mailbox-0.3/hbase/src/main/java/org/apache/james/mailbox/hbase/io/ChunkOutputStream.java Sun Sep 4 07:31:41 2011
@@ -0,0 +1,118 @@
+/****************************************************************
+ * 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.james.mailbox.hbase.io;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.client.HTable;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.util.Bytes;
+
+/**
+ * Provide an {@link OutputStream} which will write to a row. The written data
+ * will be split up by chunks of the given chunkSize. Each chunk we get written
+ * to own column which will have the chunk number (starting at 0) as column key
+ * (Long).
+ *
+ * This implementation is not thread-safe!
+ * Based on Hector implementation for Cassandra.
+ * https://github.com/rantav/hector/blob/master/core/src/main/java/me/prettyprint/cassandra/io/ChunkOutputStream.java
+ */
+public class ChunkOutputStream extends OutputStream {
+
+ private final Configuration conf;
+ private final byte[] tableName;
+ private final byte[] cf;
+ private final byte[] key;
+ private byte[] chunk;
+ private long chunkPos = 1;
+ private long pos = 0;
+
+ /**
+ * Creates a special type of {@link OutputStream} that writes data directly to HBase.
+ * @param conf HBase cluster configuration
+ * @param tableName name of the table that writes will be made
+ * @param cf name of the column family where data is going to be written
+ * @param key the row key
+ * @param chunkSize the size of each column, in bytes. For HBase, max is 10MB
+ */
+ public ChunkOutputStream(Configuration conf, byte[] tableName, byte[] cf, byte[] key, int chunkSize) {
+ this.conf = conf;
+ this.tableName = tableName;
+ this.cf = cf;
+ this.key = key;
+ this.chunk = new byte[chunkSize];
+ }
+
+ /*
+ * (non-Javadoc)
+ *
+ * @see java.io.OutputStream#write(int)
+ */
+ @Override
+ public void write(int b) throws IOException {
+ if (chunk.length - 1 == pos) {
+ flush();
+ }
+ chunk[(int) pos++] = (byte) b;
+ }
+
+ @Override
+ public void close() throws IOException {
+ writeData(true);
+ }
+
+ /**
+ * Trigger a flush. This will only write the content to the column if the
+ * chunk size is reached
+ */
+ @Override
+ public void flush() throws IOException {
+ writeData(false);
+ }
+
+ /**
+ * Write the data to column if the configured chunk size is reached or if the
+ * stream should be closed
+ *
+ * @param close
+ * @throws IOException
+ */
+ private void writeData(boolean close) throws IOException {
+ if (pos != 0 && (close || pos == chunk.length - 1)) {
+ HTable messages = null;
+ try {
+ messages = new HTable(conf, tableName);
+ Put put = new Put(key);
+ put.add(cf, Bytes.toBytes(chunkPos), Bytes.head(chunk, (int) pos + 1));
+ messages.put(put);
+ chunkPos++;
+ pos = 0;
+
+ } catch (IOException e) {
+ throw new IOException("Unable to write data", e);
+ } finally {
+ if (messages != null) {
+ messages.close();
+ }
+ }
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org