View Javadoc

    1   /***
    2    *
    3    * Copyright 2004 Hiram Chirino
    4    *
    5    * Licensed under the Apache License, Version 2.0 (the "License");
    6    * you may not use this file except in compliance with the License.
    7    * You may obtain a copy of the License at
    8    *
    9    * http://www.apache.org/licenses/LICENSE-2.0
    10   *
    11   * Unless required by applicable law or agreed to in writing, software
    12   * distributed under the License is distributed on an "AS IS" BASIS,
    13   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    14   * See the License for the specific language governing permissions and
    15   * limitations under the License.
    16   *
    17   **/
    18  package org.activeio.journal.active;
    19
    20  import java.io.DataInput;
    21  import java.io.DataInputStream;
    22  import java.io.DataOutput;
    23  import java.io.DataOutputStream;
    24  import java.io.IOException;
    25  import java.util.zip.CRC32;
    26
    27  import org.activeio.Disposable;
    28  import org.activeio.Packet;
    29  import org.activeio.adapter.PacketInputStream;
    30  import org.activeio.adapter.PacketOutputStream;
    31  import org.activeio.packet.ByteArrayPacket;
    32
    33
    34  /***
    35   * Serializes/Deserializes data records.
    36   *
    37   * @version $Revision: 1.1 $
    38   */
    39  final public class Record implements Disposable {
    40
    41      static final public int RECORD_HEADER_SIZE=8+Location.SERIALIZED_SIZE;
    42      static final public int RECORD_FOOTER_SIZE=12+Location.SERIALIZED_SIZE;
    43  	static final public int RECORD_BASE_SIZE=RECORD_HEADER_SIZE+RECORD_FOOTER_SIZE;
    44
    45      static final public byte[] START_OF_RECORD 	= new byte[] { 'S', 'o', 'R' };
    46      static final public byte[] END_OF_RECORD 	= new byte[] { 'E', 'o', 'R', '.' };
    47
    48  	static final public int SELECTED_CHECKSUM_ALGORITHIM;
    49  	static final public int NO_CHECKSUM_ALGORITHIM=0;
    50  	static final public int HASH_CHECKSUM_ALGORITHIM=1;
    51  	static final public int CRC32_CHECKSUM_ALGORITHIM=2;
    52
    53  	static {
    54  		String type = System.getProperty("org.activeio.journal.active.SELECTED_CHECKSUM_ALGORITHIM", "none");
    55  		if( "none".equals(type) ) {
    56  			SELECTED_CHECKSUM_ALGORITHIM = NO_CHECKSUM_ALGORITHIM;
    57  		} else if( "crc32".equals(type) ) {
    58  			SELECTED_CHECKSUM_ALGORITHIM = CRC32_CHECKSUM_ALGORITHIM;
    59  		} else if( "hash".equals(type) ) {
    60  			SELECTED_CHECKSUM_ALGORITHIM = HASH_CHECKSUM_ALGORITHIM;
    61  		} else {
    62  			System.err.println("System property 'org.activeio.journal.active.SELECTED_CHECKSUM_ALGORITHIM' not set properly.  Valid values are: 'none', 'hash', or 'crc32'");
    63  			SELECTED_CHECKSUM_ALGORITHIM = NO_CHECKSUM_ALGORITHIM;
    64  		}
    65  	}
    66
    67  	static public boolean isChecksumingEnabled() {
    68  		return SELECTED_CHECKSUM_ALGORITHIM!=NO_CHECKSUM_ALGORITHIM;
    69  	}
    70
    71      private final ByteArrayPacket headerFooterPacket = new ByteArrayPacket(new byte[RECORD_BASE_SIZE]);
    72      private final DataOutputStream headerFooterData = new DataOutputStream(new PacketOutputStream(headerFooterPacket));
    73
    74      private int payloadLength;
    75      private Location location;
    76      private byte recordType;
    77      private long checksum;
    78      private Location mark;
    79      private Packet payload;
    80
    81      public Record() {
    82      }
    83
    84      public Record(byte recordType, Packet payload, Location mark) throws IOException {
    85          this(null, recordType, payload, mark);
    86      }
    87
    88      public Record(Location location, byte recordType, Packet payload, Location mark) throws IOException {
    89          this.location = location;
    90          this.recordType = recordType;
    91          this.mark = mark;
    92          this.payload = payload.slice();
    93          this.payloadLength = payload.remaining();
    94          if( isChecksumingEnabled() ) {
    95              checksum(new DataInputStream(new PacketInputStream(this.payload)));
    96          }
    97
    98          writeHeader(headerFooterData);
    99          writeFooter(headerFooterData);
    100     }
    101
    102     public void setLocation(Location location) throws IOException {
    103         this.location = location;
    104         headerFooterPacket.clear();
    105         headerFooterPacket.position(8);
    106         location.writeToDataOutput(headerFooterData);
    107         headerFooterPacket.position(RECORD_HEADER_SIZE+8);
    108         location.writeToDataOutput(headerFooterData);
    109         payload.clear();
    110         headerFooterPacket.position(0);
    111         headerFooterPacket.limit(RECORD_HEADER_SIZE);
    112     }
    113
    114 	private void writeHeader( DataOutput out ) throws IOException {
    115 	    out.write(START_OF_RECORD);
    116 	    out.writeByte(recordType);
    117 	    out.writeInt(payloadLength);
    118         if( location!=null )
    119             location.writeToDataOutput(out);
    120         else
    121             out.writeLong(0);
    122 	}
    123
    124 	public void readHeader( DataInput in ) throws IOException {
    125         readAndCheckConstant(in, START_OF_RECORD, "Invalid record header: start of record constant missing.");
    126         recordType = in.readByte();
    127         payloadLength = in.readInt();
    128         if( payloadLength < 0 )
    129             throw new IOException("Invalid record header: record length cannot be less than zero.");
    130         location = Location.readFromDataInput(in);
    131 	}
    132
    133 	private void writeFooter( DataOutput out ) throws IOException {
    134 	    out.writeLong(checksum);
    135         if( location!=null )
    136             location.writeToDataOutput(out);
    137         else
    138             out.writeLong(0);
    139 	    out.write(END_OF_RECORD);
    140 	}
    141
    142 	public void readFooter( DataInput in ) throws IOException {
    143 	    long l = in.readLong();
    144         if( isChecksumingEnabled() ) {
    145             if( l!=checksum )
    146                 throw new IOException("Invalid record footer: checksum does not match.");
    147         } else {
    148             checksum = l;
    149         }
    150
    151         Location loc = Location.readFromDataInput(in);
    152         if( !loc.equals(location) )
    153             throw new IOException("Invalid record footer: location id does not match.");
    154
    155         readAndCheckConstant(in, END_OF_RECORD, "Invalid record header: end of record constant missing.");
    156 	}
    157
    158     /***
    159      * @param randomAccessFile
    160      * @throws IOException
    161      */
    162 	public void checksum(DataInput in) throws IOException {
    163 		if( SELECTED_CHECKSUM_ALGORITHIM==HASH_CHECKSUM_ALGORITHIM ) {
    164
    165 		    byte  buffer[] = new byte[1024];
    166 			byte rc[] = new byte[8];
    167 			for (int i = 0; i < payloadLength;) {
    168 			    int l = Math.min(buffer.length, payloadLength-i);
    169 				in.readFully(buffer,0,l);
    170 				for (int j = 0; j < l; j++) {
    171 					rc[j%8] ^= buffer[j];
    172 				}
    173 				i+=l;
    174 			}
    175 			checksum = (rc[0])|(rc[1]<<1)|(rc[2]<<2)|(rc[3]<<3)|(rc[4]<<4)|(rc[5]<<5)|(rc[6]<<6)|(rc[7]<<7) ;
    176
    177 		} else if( SELECTED_CHECKSUM_ALGORITHIM==CRC32_CHECKSUM_ALGORITHIM ) {
    178 			byte  buffer[] = new byte[1024];
    179 			CRC32 crc32 = new CRC32();
    180 			for (int i = 0; i < payloadLength;) {
    181 			    int l = Math.min(buffer.length, payloadLength-i);
    182 				in.readFully(buffer,0,l);
    183 				crc32.update(buffer,0,l);
    184 				i+=l;
    185 			}
    186 			checksum = crc32.getValue();
    187 		} else {
    188 		    checksum = 0L;
    189 		}
    190     }
    191
    192
    193     /***
    194      */
    195     private void readAndCheckConstant(DataInput in, byte[] byteConstant, String errorMessage ) throws IOException {
    196         for (int i = 0; i < byteConstant.length; i++) {
    197             byte checkByte = byteConstant[i];
    198             if( in.readByte()!= checkByte ) {
    199                 throw new IOException(errorMessage);
    200             }
    201         }
    202     }
    203
    204     public boolean readFromPacket(Packet packet) throws IOException {
    205         Packet dup = packet.duplicate();
    206
    207         if( dup.remaining() < RECORD_HEADER_SIZE )
    208             return false;
    209         DataInputStream is = new DataInputStream(new PacketInputStream(dup));
    210         readHeader( is );
    211         if( dup.remaining() < payloadLength+RECORD_FOOTER_SIZE ) {
    212             return false;
    213         }
    214
    215         // Set limit to create a slice of the payload.
    216         dup.limit(dup.position()+payloadLength);
    217         this.payload = dup.slice();
    218 	    if( isChecksumingEnabled() ) {
    219 	        checksum(new DataInputStream(new PacketInputStream(payload)));
    220 	    }
    221
    222 	    // restore the limit and seek to the footer.
    223         dup.limit(packet.limit());
    224         dup.position(dup.position()+payloadLength);
    225         readFooter(is);
    226
    227         // If every thing went well.. advance the position of the orignal packet.
    228         packet.position(dup.position());
    229         dup.dispose();
    230         return true;
    231     }
    232
    233     /***
    234      * @return Returns the checksum.
    235      */
    236     public long getChecksum() {
    237         return checksum;
    238     }
    239
    240     /***
    241      * @return Returns the length.
    242      */
    243     public int getPayloadLength() {
    244         return payloadLength;
    245     }
    246
    247     /***
    248      * @return Returns the length of the record .
    249      */
    250     public int getRecordLength() {
    251         return payloadLength+Record.RECORD_BASE_SIZE;
    252     }
    253
    254     /***
    255      * @return Returns the location.
    256      */
    257     public Location getLocation() {
    258         return location;
    259     }
    260
    261     /***
    262      * @return Returns the mark.
    263      */
    264     public Location getMark() {
    265         return mark;
    266     }
    267
    268     /***
    269      * @return Returns the payload.
    270      */
    271     public Packet getPayload() {
    272         return payload;
    273     }
    274
    275     /***
    276      * @return Returns the recordType.
    277      */
    278     public byte getRecordType() {
    279         return recordType;
    280     }
    281
    282 	public boolean hasRemaining() {
    283 		return headerFooterPacket.position()!=RECORD_BASE_SIZE;
    284 	}
    285
    286 	public void read(Packet packet) {
    287
    288 		// push the header
    289 		headerFooterPacket.read(packet);
    290 		// push the payload.
    291 		payload.read(packet);
    292
    293 		// Can we switch to the footer now?
    294 		if( !payload.hasRemaining() && headerFooterPacket.position()==RECORD_HEADER_SIZE ) {
    295 			headerFooterPacket.position(RECORD_HEADER_SIZE);
    296              headerFooterPacket.limit(RECORD_BASE_SIZE);
    297 			headerFooterPacket.read(packet);
    298 		}
    299
    300 	}
    301
    302     public void dispose() {
    303         if( payload!=null ) {
    304             payload.dispose();
    305             payload=null;
    306         }
    307     }
    308
    309 }