path: root/src/main/java/org/openslx/imagemaster/db/mappers/DbImageBlock.java

package org.openslx.imagemaster.db.mappers;

import java.nio.ByteBuffer;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.openslx.filetransfer.FileRange;
import org.openslx.filetransfer.util.ChunkStatus;
import org.openslx.filetransfer.util.FileChunk;
import org.openslx.imagemaster.db.Database;
import org.openslx.imagemaster.db.MysqlConnection;
import org.openslx.imagemaster.db.MysqlStatement;

public class DbImageBlock
{

	private static final Logger LOGGER = LogManager.getLogger( DbImageBlock.class );

	private static AsyncThread asyncBlockUpdate = null;

	private static synchronized void initAsyncThread()
	{
		if ( asyncBlockUpdate == null ) {
			asyncBlockUpdate = new AsyncThread();
			asyncBlockUpdate.start();
		}
	}

	public static void asyncUpdate( String imageVersionId, FileChunk chunk ) throws InterruptedException
	{
		initAsyncThread();
		asyncBlockUpdate.put( new ChunkUpdate( imageVersionId, chunk.range, chunk.getStatus() != ChunkStatus.COMPLETE ) );
	}

	private static class AsyncThread extends Thread
	{
		private final ArrayBlockingQueue<ChunkUpdate> queue = new ArrayBlockingQueue<>( 100 );

		public void put( ChunkUpdate chunk ) throws InterruptedException
		{
			queue.put( chunk );
		}

		@Override
		public void run()
		{
			try {
				while ( !interrupted() ) {
					ChunkUpdate chunk = queue.take();
					Thread.sleep( 100 );
					try ( MysqlConnection connection = Database.getConnection() ) {
						MysqlStatement stmt = connection.prepareStatement( "UPDATE imageblock SET ismissing = :ismissing"
								+ " WHERE imageversionid = :imageversionid AND startbyte = :startbyte AND blocksize = :blocksize" );
						do {
							stmt.setBoolean( "ismissing", chunk.isMissing );
							stmt.setString( "imageversionid", chunk.imageVersionId );
							stmt.setLong( "startbyte", chunk.range.startOffset );
							stmt.setInt( "blocksize", chunk.range.getLength() );
							stmt.executeUpdate();
							chunk = queue.poll();
						} while ( chunk != null );
						connection.commit();
					} catch ( SQLException e ) {
						LOGGER.error( "Query failed in DbImageBlock.AsyncThread.run()", e );
						continue;
					}
					Thread.sleep( 2000 );
				}
			} catch ( InterruptedException e ) {
				LOGGER.debug( "async thread interrupted" );
				interrupt();
			}
		}
	}

	private static class ChunkUpdate
	{
		public final String imageVersionId;
		public final FileRange range;
		public final boolean isMissing;

		public ChunkUpdate( String imageVersionId, FileRange range, boolean isMissing )
		{
			this.imageVersionId = imageVersionId;
			this.range = range;
			this.isMissing = isMissing;
		}
	}

	public static void insertChunkList( String imageVersionId, List<FileChunk> all, boolean missing ) throws SQLException
	{
		try ( MysqlConnection connection = Database.getConnection() ) {
			MysqlStatement stmt = connection.prepareStatement( "INSERT IGNORE INTO imageblock"
					+ " (imageversionid, startbyte, blocksize, blocksha1, ismissing) VALUES"
					+ " (:imageversionid, :startbyte, :blocksize, :blocksha1, :ismissing)" );
			stmt.setString( "imageversionid", imageVersionId );
			stmt.setBoolean( "ismissing", missing );
			for ( FileChunk chunk : all ) {
				stmt.setLong( "startbyte", chunk.range.startOffset );
				stmt.setInt( "blocksize", chunk.range.getLength() );
				stmt.setBinary( "blocksha1", chunk.getSha1Sum() );
				stmt.executeUpdate();
			}
			connection.commit();
		} catch ( SQLException e ) {
			LOGGER.error( "Query failed in DbImageBlock.insertChunkList()", e );
			throw e;
		}
	}

	public static List<Boolean> getMissingStatusList( String imageVersionId ) throws SQLException
	{
		try ( MysqlConnection connection = Database.getConnection() ) {
			MysqlStatement stmt = connection.prepareStatement( "SELECT startbyte, ismissing FROM imageblock"
					+ " WHERE imageversionid = :imageversionid ORDER BY startbyte ASC" );
			stmt.setString( "imageversionid", imageVersionId );
			ResultSet rs = stmt.executeQuery();
			List<Boolean> list = new ArrayList<>();
			long expectedOffset = 0;
			while ( rs.next() ) {
				long currentOffset = rs.getLong( "startbyte" );
				if ( currentOffset < expectedOffset )
					continue;
				while ( currentOffset > expectedOffset ) {
					list.add( Boolean.TRUE );
					expectedOffset += FileChunk.CHUNK_SIZE;
				}
				if ( currentOffset == expectedOffset ) {
					list.add( rs.getBoolean( "ismissing" ) );
					expectedOffset += FileChunk.CHUNK_SIZE;
				}
			}
			return list;
		} catch ( SQLException e ) {
			LOGGER.error( "Query failed in DbImageBlock.getBlockStatuses()", e );
			throw e;
		}
	}

	public static List<ByteBuffer> getBlockHashes( String imageVersionId ) throws SQLException
	{
		try ( MysqlConnection connection = Database.getConnection() ) {
			return getBlockHashes( connection, imageVersionId );
		} catch ( SQLException e ) {
			LOGGER.error( "Query failed in DbImageBlock.getBlockHashes()", e );
			throw e;
		}
	}

	private static List<ByteBuffer> getBlockHashes( MysqlConnection connection, String imageVersionId )
			throws SQLException
	{
		MysqlStatement stmt = connection.prepareStatement( "SELECT startbyte, blocksha1 FROM imageblock"
				+ " WHERE imageversionid = :imageversionid ORDER BY startbyte ASC" );
		stmt.setString( "imageversionid", imageVersionId );
		ResultSet rs = stmt.executeQuery();
		List<ByteBuffer> list = new ArrayList<>();
		long expectedOffset = 0;
		while ( rs.next() ) {
			long currentOffset = rs.getLong( "startbyte" );
			if ( currentOffset < expectedOffset )
				continue;
			while ( currentOffset > expectedOffset ) {
				list.add( null );
				expectedOffset += FileChunk.CHUNK_SIZE;
			}
			if ( currentOffset == expectedOffset ) {
				list.add( ByteBuffer.wrap( rs.getBytes( "blocksha1" ) ) );
				expectedOffset += FileChunk.CHUNK_SIZE;
			}
		}
		return list;
	}

}