core/org.eclipse.smila.jdbc.test/code/src/org/eclipse/smila/jdbc/test/TestJdbcWriterService.java - gerrit/smila/org.eclipse.smila.core - Git at Google

 /*********************************************************************************************************************
  * Copyright (c) 2008, 2013 Empolis Information Management GmbH and brox IT Solutions GmbH. All rights reserved. This
  * program and the accompanying materials are made available under the terms of the Eclipse Public License v1.0 which
  * accompanies this distribution, and is available at http://www.eclipse.org/legal/epl-v10.html
  *********************************************************************************************************************/
 package org.eclipse.smila.jdbc.test;

 import java.sql.Connection;
 import java.sql.SQLException;
 import java.sql.Statement;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;

 import org.eclipse.smila.common.logging.MessageCollector;
 import org.eclipse.smila.datamodel.AnyMap;
 import org.eclipse.smila.datamodel.DataFactory;
 import org.eclipse.smila.datamodel.Record;
 import org.eclipse.smila.datamodel.Value;
 import org.eclipse.smila.jdbc.JdbcProvider;
 import org.eclipse.smila.jdbc.JdbcWriterService;
 import org.eclipse.smila.jdbc.JdbcWriterServiceException;
 import org.eclipse.smila.jdbc.SqlExecutor;
 import org.eclipse.smila.jdbc.internal.JdbcWriterServiceImpl;
 import org.eclipse.smila.test.DeclarativeServiceTestCase;

 /** Test for {@link TestJdbcWriterService} class. */
 public class TestJdbcWriterService extends DeclarativeServiceTestCase {

   private static final String DB_URL1 = "jdbc:derby:memory:testwriter1";

   private static final String DB_URL2 = "jdbc:derby:memory:testwriter2";

   private static final List<Value> NO_VALUES = Collections.emptyList();

   private static final int WAIT_SECONDS = 2;

   /** service under test */
   private JdbcWriterService _writer;

   private JdbcProvider _provider;

   private final AnyMap _dbProps = DataFactory.DEFAULT.createAnyMap();

   private Connection _conn1;

   private Connection _conn2;

   @Override
   protected void setUp() throws Exception {
     super.setUp();
     System.out.println("setUp " + getName());
     _writer = getService(JdbcWriterService.class);
     _provider = getService(JdbcProvider.class);
     _dbProps.put(JdbcWriterService.DB_PROPERTY_USER_NAME, "user");
     _dbProps.put(JdbcWriterService.DB_PROPERTY_USER_PASSWORD, "password");

     _conn1 = prepareConnection(_provider, DB_URL1);
     _conn2 = prepareConnection(_provider, DB_URL2);

     try (final Statement stmt = _conn1.createStatement()) {
       try {
         stmt.execute("create procedure sleep(in time bigint) parameter style java no sql language java "
           + "external name 'java.lang.Thread.sleep'");
       } catch (final SQLException ex) {
         ; // ex.printStackTrace();
       }
     }
   }

   @Override
   protected void tearDown() throws Exception {
     _conn1.close();
     _conn2.close();
     super.tearDown();
   }

   /** tests OSGI service. */
   public void testService() throws Exception {
     assertNotNull(_writer);
     assertTrue(_writer instanceof JdbcWriterServiceImpl);
   }

   public void testWriteString() throws Exception {
     final List<Value> values = asListOfValues("hello", "world");
     _writer.write(DB_URL1, _dbProps, "insert into entries (string1, string2) values (?, ?)", values);
     checkStringEntry(_conn1, "hello", "world");
   }

   public void testWriteWithoutValues() throws Exception {
     _writer.write(DB_URL1, _dbProps, "insert into entries (string1, string2) values ('jekyll', 'hyde')", null);
     checkStringEntry(_conn1, "jekyll", "hyde");
   }

   public void testWriteWithEmptyValues() throws Exception {
     _writer.write(DB_URL1, _dbProps, "insert into entries (string1, string2) values ('holmes', 'watson')",
       NO_VALUES);
     checkStringEntry(_conn1, "holmes", "watson");
   }

   public void testWriteWithNullString() throws Exception {
     final List<Value> values = asListOfValues("hello", null);
     _writer.write(DB_URL1, _dbProps, "insert into entries (string1, string2) values (?, ?)", values);
     checkStringEntry(_conn1, "hello", null);
   }

   public void testWriteWithNullInteger() throws Exception {
     final List<Value> values = asListOfValues("hello", null);
     _writer.write(DB_URL1, _dbProps, "insert into entries (string1, int1) values (?, ?)", values);
     final Collection<Record> entries = getEntries(1, _conn1, "select id, string1, int1 from entries order by id");
     final AnyMap entry = entries.iterator().next().getMetadata();
     assertEquals(2, entry.size());
     assertEquals("hello", entry.getStringValue("STRING1"));
     assertNull(entry.getStringValue("INT1"));
   }

   public void testCheckFifo() throws Exception {
     final int numberOfEntries = 100;
     for (int i = 0; i < numberOfEntries; i++) {
       final List<Value> values = asListOfValues(i);
       _writer.write(DB_URL1, _dbProps, "insert into entries (int1) values (?)", values);
     }
     checkIntegerEntries(_conn1, numberOfEntries);
   }

   public void testTwoConnections() throws Exception {
     final List<Value> values1 = asListOfValues("hello", "world 1");
     _writer.write(DB_URL1, _dbProps, "insert into entries (string1, string2) values (?, ?)", values1);
     final List<Value> values2 = asListOfValues("hello, too", "world 2");
     _writer.write(DB_URL2, _dbProps, "insert into entries (string1, string2) values (?, ?)", values2);

     checkStringEntry(_conn1, "hello", "world 1");
     checkStringEntry(_conn2, "hello, too", "world 2");
   }

   public void testNoErrorOnInvalidConnection() throws Exception {
     final List<Value> values = asListOfValues("hello", "world");
     // On Windows, this makes following tests fails, because connection management in JdbcWriterService blocks.
     // Should be improved.
     // _writer.write("jdbc:derby://nosuchserver:1527/nosuchdb", _dbProps,
     _writer.write("jdbc:derby:", _dbProps,
       "insert into entries (string1, string2) values (?, ?)", values);
   }

   public void testNoErrorOnInvalidStatement() throws Exception {
     final List<Value> values = asListOfValues("hello", "world");
     _writer.write(DB_URL1, _dbProps, "this is not an SQL statement", values);
     getEntries(0, _conn1, "select * from entries");
   }

   public void testErrorOnFullQueue() throws Exception {
     int numberOfEntries = 0;
     try {
       for (numberOfEntries = 0; numberOfEntries < 10000; numberOfEntries++) {
         final List<Value> values = asListOfValues(numberOfEntries);
         _writer.write(DB_URL1, _dbProps, "insert into entries (int1) values (?)", values);
       }
       fail("should not work");
     } catch (final JdbcWriterServiceException ex) {
       ; // OK
     }
     assertTrue(numberOfEntries >= 100); // default queue length
     checkIntegerEntries(_conn1, numberOfEntries);
   }

   public void testWriteIsNonBlocking() throws Exception {
     final List<Value> values = asListOfValues("hello", "world");
     final int blockTime = WAIT_SECONDS * 1000 + 200;
     _writer.write(DB_URL1, _dbProps, "call sleep(" + blockTime + ")", NO_VALUES);
     _writer.write(DB_URL1, _dbProps, "insert into entries (string1, string2) values (?, ?)", values);
     final long start = System.currentTimeMillis();
     try {
       checkStringEntry(_conn1, "hello", "world");
       fail("should not work");
     } catch (final Throwable ex) {
       ; // OK
     }
     assertTrue(System.currentTimeMillis() - start >= WAIT_SECONDS * 1000);
     Thread.sleep(200);
   }

   public void testBlockedConnectionDoesNotBlockOtherConnection() throws Exception {
     final List<Value> values = asListOfValues("hello", "world");
     final int blockTime = WAIT_SECONDS * 1000;
     _writer.write(DB_URL1, _dbProps, "call sleep(" + blockTime + ")", NO_VALUES);
     _writer.write(DB_URL2, _dbProps, "insert into entries (string1, string2) values (?, ?)", values);
     final long start = System.currentTimeMillis();
     checkStringEntry(_conn2, "hello", "world");
     assertTrue(System.currentTimeMillis() - start < blockTime / 2);
     Thread.sleep(blockTime);
   }

   public static Connection prepareConnection(final JdbcProvider provider, final String url) throws SQLException {
     final Connection conn = provider.getConnection(url + ";create=true", "user", "password");
     try (final Statement stmt = conn.createStatement()) {
       try {
         stmt.execute("drop table entries");
       } catch (final Exception ex) {
         ; // table didn't exist
       }
       stmt.execute("create table entries (id int generated always as identity, "
         + "string1 varchar(255), string2 varchar(255), int1 int)");
       conn.commit();
       return conn;
     }
   }

   public static void checkStringEntry(final Connection conn, final String string1, final String string2)
     throws Exception {
     final Collection<Record> entries = getEntries(1, conn, "select id, string1, string2 from entries order by id");
     final AnyMap entry = entries.iterator().next().getMetadata();
     int expectedAttributeCount = 1;
     if (string1 != null) {
       expectedAttributeCount++;
     }
     if (string2 != null) {
       expectedAttributeCount++;
     }
     assertEquals(expectedAttributeCount, entry.size());
     assertEquals(string1, entry.getStringValue("STRING1"));
     assertEquals(string2, entry.getStringValue("STRING2"));
   }

   public static void checkIntegerEntries(final Connection conn, final int numberOfEntries) throws Exception {
     final Collection<Record> entries =
       getEntries(numberOfEntries, conn, "select id, int1 from entries order by id");
     final Iterator<Record> iter = entries.iterator();
     for (int i = 0; i < numberOfEntries; i++) {
       final AnyMap entry = iter.next().getMetadata();
       assertEquals(2, entry.size());
       assertEquals(i, entry.getLongValue("INT1").intValue());
     }
   }

   public static Collection<Record> getEntries(final int expectedNumberOfEntries, final Connection conn,
     final String sql) throws Exception {
     try (final SqlExecutor exec = new SqlExecutor(conn, sql, 0, new MessageCollector.Ignore())) {
       for (int i = 0; i < WAIT_SECONDS * 10; i++) {
         final Collection<Record> entries = exec.execute(NO_VALUES);
         if (entries.size() >= expectedNumberOfEntries) {
           assertEquals(expectedNumberOfEntries, entries.size());
           return entries;
         }
         Thread.sleep(100);
       }
       fail("Did not get " + expectedNumberOfEntries + " entries within " + WAIT_SECONDS + " seconds.");
       return null;
     }
   }

   private List<Value> asListOfValues(final Object... values) {
     final List<Value> anyValues = new ArrayList<>(values.length);
     for (final Object value : values) {
       if (value == null) {
         anyValues.add(null);
       } else {
         anyValues.add(DataFactory.DEFAULT.autoConvertValue(value));
       }
     }
     return anyValues;

   }
 }
	/*********************************************************************************************************************
	* Copyright (c) 2008, 2013 Empolis Information Management GmbH and brox IT Solutions GmbH. All rights reserved. This
	* program and the accompanying materials are made available under the terms of the Eclipse Public License v1.0 which
	* accompanies this distribution, and is available at http://www.eclipse.org/legal/epl-v10.html
	*********************************************************************************************************************/
	package org.eclipse.smila.jdbc.test;

	import java.sql.Connection;
	import java.sql.SQLException;
	import java.sql.Statement;
	import java.util.ArrayList;
	import java.util.Collection;
	import java.util.Collections;
	import java.util.Iterator;
	import java.util.List;

	import org.eclipse.smila.common.logging.MessageCollector;
	import org.eclipse.smila.datamodel.AnyMap;
	import org.eclipse.smila.datamodel.DataFactory;
	import org.eclipse.smila.datamodel.Record;
	import org.eclipse.smila.datamodel.Value;
	import org.eclipse.smila.jdbc.JdbcProvider;
	import org.eclipse.smila.jdbc.JdbcWriterService;
	import org.eclipse.smila.jdbc.JdbcWriterServiceException;
	import org.eclipse.smila.jdbc.SqlExecutor;
	import org.eclipse.smila.jdbc.internal.JdbcWriterServiceImpl;
	import org.eclipse.smila.test.DeclarativeServiceTestCase;

	/** Test for {@link TestJdbcWriterService} class. */
	public class TestJdbcWriterService extends DeclarativeServiceTestCase {

	private static final String DB_URL1 = "jdbc:derby:memory:testwriter1";

	private static final String DB_URL2 = "jdbc:derby:memory:testwriter2";

	private static final List<Value> NO_VALUES = Collections.emptyList();

	private static final int WAIT_SECONDS = 2;

	/** service under test */
	private JdbcWriterService _writer;

	private JdbcProvider _provider;

	private final AnyMap _dbProps = DataFactory.DEFAULT.createAnyMap();

	private Connection _conn1;

	private Connection _conn2;

	@Override
	protected void setUp() throws Exception {
	super.setUp();
	System.out.println("setUp " + getName());
	_writer = getService(JdbcWriterService.class);
	_provider = getService(JdbcProvider.class);
	_dbProps.put(JdbcWriterService.DB_PROPERTY_USER_NAME, "user");
	_dbProps.put(JdbcWriterService.DB_PROPERTY_USER_PASSWORD, "password");

	_conn1 = prepareConnection(_provider, DB_URL1);
	_conn2 = prepareConnection(_provider, DB_URL2);

	try (final Statement stmt = _conn1.createStatement()) {
	try {
	stmt.execute("create procedure sleep(in time bigint) parameter style java no sql language java "
	+ "external name 'java.lang.Thread.sleep'");
	} catch (final SQLException ex) {
	; // ex.printStackTrace();
	}
	}
	}

	@Override
	protected void tearDown() throws Exception {
	_conn1.close();
	_conn2.close();
	super.tearDown();
	}

	/** tests OSGI service. */
	public void testService() throws Exception {
	assertNotNull(_writer);
	assertTrue(_writer instanceof JdbcWriterServiceImpl);
	}

	public void testWriteString() throws Exception {
	final List<Value> values = asListOfValues("hello", "world");
	_writer.write(DB_URL1, _dbProps, "insert into entries (string1, string2) values (?, ?)", values);
	checkStringEntry(_conn1, "hello", "world");
	}

	public void testWriteWithoutValues() throws Exception {
	_writer.write(DB_URL1, _dbProps, "insert into entries (string1, string2) values ('jekyll', 'hyde')", null);
	checkStringEntry(_conn1, "jekyll", "hyde");
	}

	public void testWriteWithEmptyValues() throws Exception {
	_writer.write(DB_URL1, _dbProps, "insert into entries (string1, string2) values ('holmes', 'watson')",
	NO_VALUES);
	checkStringEntry(_conn1, "holmes", "watson");
	}

	public void testWriteWithNullString() throws Exception {
	final List<Value> values = asListOfValues("hello", null);
	_writer.write(DB_URL1, _dbProps, "insert into entries (string1, string2) values (?, ?)", values);
	checkStringEntry(_conn1, "hello", null);
	}

	public void testWriteWithNullInteger() throws Exception {
	final List<Value> values = asListOfValues("hello", null);
	_writer.write(DB_URL1, _dbProps, "insert into entries (string1, int1) values (?, ?)", values);
	final Collection<Record> entries = getEntries(1, _conn1, "select id, string1, int1 from entries order by id");
	final AnyMap entry = entries.iterator().next().getMetadata();
	assertEquals(2, entry.size());
	assertEquals("hello", entry.getStringValue("STRING1"));
	assertNull(entry.getStringValue("INT1"));
	}

	public void testCheckFifo() throws Exception {
	final int numberOfEntries = 100;
	for (int i = 0; i < numberOfEntries; i++) {
	final List<Value> values = asListOfValues(i);
	_writer.write(DB_URL1, _dbProps, "insert into entries (int1) values (?)", values);
	}
	checkIntegerEntries(_conn1, numberOfEntries);
	}

	public void testTwoConnections() throws Exception {
	final List<Value> values1 = asListOfValues("hello", "world 1");
	_writer.write(DB_URL1, _dbProps, "insert into entries (string1, string2) values (?, ?)", values1);
	final List<Value> values2 = asListOfValues("hello, too", "world 2");
	_writer.write(DB_URL2, _dbProps, "insert into entries (string1, string2) values (?, ?)", values2);

	checkStringEntry(_conn1, "hello", "world 1");
	checkStringEntry(_conn2, "hello, too", "world 2");
	}

	public void testNoErrorOnInvalidConnection() throws Exception {
	final List<Value> values = asListOfValues("hello", "world");
	// On Windows, this makes following tests fails, because connection management in JdbcWriterService blocks.
	// Should be improved.
	// _writer.write("jdbc:derby://nosuchserver:1527/nosuchdb", _dbProps,
	_writer.write("jdbc:derby:", _dbProps,
	"insert into entries (string1, string2) values (?, ?)", values);
	}

	public void testNoErrorOnInvalidStatement() throws Exception {
	final List<Value> values = asListOfValues("hello", "world");
	_writer.write(DB_URL1, _dbProps, "this is not an SQL statement", values);
	getEntries(0, _conn1, "select * from entries");
	}

	public void testErrorOnFullQueue() throws Exception {
	int numberOfEntries = 0;
	try {
	for (numberOfEntries = 0; numberOfEntries < 10000; numberOfEntries++) {
	final List<Value> values = asListOfValues(numberOfEntries);
	_writer.write(DB_URL1, _dbProps, "insert into entries (int1) values (?)", values);
	}
	fail("should not work");
	} catch (final JdbcWriterServiceException ex) {
	; // OK
	}
	assertTrue(numberOfEntries >= 100); // default queue length
	checkIntegerEntries(_conn1, numberOfEntries);
	}

	public void testWriteIsNonBlocking() throws Exception {
	final List<Value> values = asListOfValues("hello", "world");
	final int blockTime = WAIT_SECONDS * 1000 + 200;
	_writer.write(DB_URL1, _dbProps, "call sleep(" + blockTime + ")", NO_VALUES);
	_writer.write(DB_URL1, _dbProps, "insert into entries (string1, string2) values (?, ?)", values);
	final long start = System.currentTimeMillis();
	try {
	checkStringEntry(_conn1, "hello", "world");
	fail("should not work");
	} catch (final Throwable ex) {
	; // OK
	}
	assertTrue(System.currentTimeMillis() - start >= WAIT_SECONDS * 1000);
	Thread.sleep(200);
	}

	public void testBlockedConnectionDoesNotBlockOtherConnection() throws Exception {
	final List<Value> values = asListOfValues("hello", "world");
	final int blockTime = WAIT_SECONDS * 1000;
	_writer.write(DB_URL1, _dbProps, "call sleep(" + blockTime + ")", NO_VALUES);
	_writer.write(DB_URL2, _dbProps, "insert into entries (string1, string2) values (?, ?)", values);
	final long start = System.currentTimeMillis();
	checkStringEntry(_conn2, "hello", "world");
	assertTrue(System.currentTimeMillis() - start < blockTime / 2);
	Thread.sleep(blockTime);
	}

	public static Connection prepareConnection(final JdbcProvider provider, final String url) throws SQLException {
	final Connection conn = provider.getConnection(url + ";create=true", "user", "password");
	try (final Statement stmt = conn.createStatement()) {
	try {
	stmt.execute("drop table entries");
	} catch (final Exception ex) {
	; // table didn't exist
	}
	stmt.execute("create table entries (id int generated always as identity, "
	+ "string1 varchar(255), string2 varchar(255), int1 int)");
	conn.commit();
	return conn;
	}
	}

	public static void checkStringEntry(final Connection conn, final String string1, final String string2)
	throws Exception {
	final Collection<Record> entries = getEntries(1, conn, "select id, string1, string2 from entries order by id");
	final AnyMap entry = entries.iterator().next().getMetadata();
	int expectedAttributeCount = 1;
	if (string1 != null) {
	expectedAttributeCount++;
	}
	if (string2 != null) {
	expectedAttributeCount++;
	}
	assertEquals(expectedAttributeCount, entry.size());
	assertEquals(string1, entry.getStringValue("STRING1"));
	assertEquals(string2, entry.getStringValue("STRING2"));
	}

	public static void checkIntegerEntries(final Connection conn, final int numberOfEntries) throws Exception {
	final Collection<Record> entries =
	getEntries(numberOfEntries, conn, "select id, int1 from entries order by id");
	final Iterator<Record> iter = entries.iterator();
	for (int i = 0; i < numberOfEntries; i++) {
	final AnyMap entry = iter.next().getMetadata();
	assertEquals(2, entry.size());
	assertEquals(i, entry.getLongValue("INT1").intValue());
	}
	}

	public static Collection<Record> getEntries(final int expectedNumberOfEntries, final Connection conn,
	final String sql) throws Exception {
	try (final SqlExecutor exec = new SqlExecutor(conn, sql, 0, new MessageCollector.Ignore())) {
	for (int i = 0; i < WAIT_SECONDS * 10; i++) {
	final Collection<Record> entries = exec.execute(NO_VALUES);
	if (entries.size() >= expectedNumberOfEntries) {
	assertEquals(expectedNumberOfEntries, entries.size());
	return entries;
	}
	Thread.sleep(100);
	}
	fail("Did not get " + expectedNumberOfEntries + " entries within " + WAIT_SECONDS + " seconds.");
	return null;
	}
	}

	private List<Value> asListOfValues(final Object... values) {
	final List<Value> anyValues = new ArrayList<>(values.length);
	for (final Object value : values) {
	if (value == null) {
	anyValues.add(null);
	} else {
	anyValues.add(DataFactory.DEFAULT.autoConvertValue(value));
	}
	}
	return anyValues;

	}
	}