@Test public void testUpdateMessageAttachment1() throws Exception { String channelId = "testchannel"; long messageId = 1; Attachment attachment = new Attachment("testattachment", "testing".getBytes(), "text/plain"); ResultSet segmentCountResult = mock(ResultSet.class); PreparedStatement segmentCountStatement = mock(PreparedStatement.class); PreparedStatement updateStatement = mock(PreparedStatement.class); PreparedStatement insertStatement = mock(PreparedStatement.class); PreparedStatement deleteStatement = mock(PreparedStatement.class); JdbcDao dao = getDao(channelId, segmentCountResult, segmentCountStatement, updateStatement, insertStatement, deleteStatement); when(segmentCountResult.getInt(1)).thenReturn(0); dao.updateMessageAttachment(channelId, messageId, attachment); verify(segmentCountStatement, times(1)).setString(1, attachment.getId()); verify(segmentCountStatement, times(1)).setLong(2, messageId); verify(segmentCountStatement, times(1)).executeQuery(); verify(updateStatement, times(1)).setString(1, attachment.getType()); verify(updateStatement, times(1)).setString(4, attachment.getId()); verify(updateStatement, times(1)).setLong(5, messageId); verify(insertStatement, times(1)).setString(1, attachment.getId()); verify(insertStatement, times(1)).setLong(2, messageId); verify(insertStatement, times(1)).setString(3, attachment.getType()); verify(insertStatement, times(1)).setInt(4, 1); verify(insertStatement, times(1)).setInt(5, attachment.getContent().length); verify(insertStatement, times(1)).setBytes(6, attachment.getContent()); verify(insertStatement, times(1)).executeUpdate(); verify(updateStatement, times(1)).clearParameters(); verify(insertStatement, times(1)).clearParameters(); verify(deleteStatement, times(0)).setString(1, attachment.getId()); verify(deleteStatement, times(0)).setLong(2, messageId); verify(deleteStatement, times(0)).setInt(eq(3), anyInt()); verify(deleteStatement, times(0)).executeUpdate(); }
@Override public void updateMessageAttachment(String channelId, long messageId, Attachment attachment) { logger.debug(channelId + "/" + messageId + ": updating message attachment"); PreparedStatement segmentCountStatement = null; ResultSet segmentCountResult = null; PreparedStatement updateStatement = null; PreparedStatement insertStatement = null; try { segmentCountStatement = prepareStatement("selectMessageAttachmentSegmentCount", channelId); segmentCountStatement.setString(1, attachment.getId()); segmentCountStatement.setLong(2, messageId); segmentCountResult = segmentCountStatement.executeQuery(); segmentCountResult.next(); int totalSegmentCount = segmentCountResult.getInt(1); updateStatement = prepareStatement("updateMessageAttachment", channelId); updateStatement.setString(1, attachment.getType()); updateStatement.setString(4, attachment.getId()); updateStatement.setLong(5, messageId); insertStatement = prepareStatement("insertMessageAttachment", channelId); insertStatement.setString(1, attachment.getId()); insertStatement.setLong(2, messageId); insertStatement.setString(3, attachment.getType()); int chunkSize = 10000000; ByteArrayInputStream inputStream = null; boolean done = false; int currentSegmentId = 1; while (!done) { PreparedStatement statement; int segmentIdIndex; int attachmentSizeIndex; int contentIndex; if (currentSegmentId <= totalSegmentCount) { statement = updateStatement; segmentIdIndex = 6; attachmentSizeIndex = 2; contentIndex = 3; } else { statement = insertStatement; segmentIdIndex = 4; attachmentSizeIndex = 5; contentIndex = 6; } statement.setInt(segmentIdIndex, currentSegmentId); if (attachment.getContent().length <= chunkSize) { statement.setInt(attachmentSizeIndex, attachment.getContent().length); statement.setBytes(contentIndex, attachment.getContent()); statement.executeUpdate(); currentSegmentId++; done = true; } else { if (inputStream == null) { inputStream = new ByteArrayInputStream(attachment.getContent()); } if (inputStream.available() > 0) { int segmentSize = Math.min(chunkSize, inputStream.available()); byte[] segment = new byte[segmentSize]; inputStream.read(segment, 0, segmentSize); statement.setInt(attachmentSizeIndex, segmentSize); statement.setBytes(contentIndex, segment); statement.executeUpdate(); currentSegmentId++; } else { done = true; } } } updateStatement.clearParameters(); insertStatement.clearParameters(); if (totalSegmentCount >= currentSegmentId) { PreparedStatement deleteStatement = prepareStatement("deleteMessageAttachmentLingeringSegments", channelId); deleteStatement.setString(1, attachment.getId()); deleteStatement.setLong(2, messageId); deleteStatement.setInt(3, currentSegmentId); deleteStatement.executeUpdate(); } } catch (SQLException e) { throw new DonkeyDaoException(e); } finally { List<AutoCloseable> dbObjects = new ArrayList<>(); dbObjects.add(segmentCountStatement); dbObjects.add(segmentCountResult); dbObjects.add(updateStatement); dbObjects.add(insertStatement); closeDatabaseObjectsIfNeeded(dbObjects); } }
JdbcDao implements DonkeyDao { @Override public void updateMessageAttachment(String channelId, long messageId, Attachment attachment) { logger.debug(channelId + "/" + messageId + ": updating message attachment"); PreparedStatement segmentCountStatement = null; ResultSet segmentCountResult = null; PreparedStatement updateStatement = null; PreparedStatement insertStatement = null; try { segmentCountStatement = prepareStatement("selectMessageAttachmentSegmentCount", channelId); segmentCountStatement.setString(1, attachment.getId()); segmentCountStatement.setLong(2, messageId); segmentCountResult = segmentCountStatement.executeQuery(); segmentCountResult.next(); int totalSegmentCount = segmentCountResult.getInt(1); updateStatement = prepareStatement("updateMessageAttachment", channelId); updateStatement.setString(1, attachment.getType()); updateStatement.setString(4, attachment.getId()); updateStatement.setLong(5, messageId); insertStatement = prepareStatement("insertMessageAttachment", channelId); insertStatement.setString(1, attachment.getId()); insertStatement.setLong(2, messageId); insertStatement.setString(3, attachment.getType()); int chunkSize = 10000000; ByteArrayInputStream inputStream = null; boolean done = false; int currentSegmentId = 1; while (!done) { PreparedStatement statement; int segmentIdIndex; int attachmentSizeIndex; int contentIndex; if (currentSegmentId <= totalSegmentCount) { statement = updateStatement; segmentIdIndex = 6; attachmentSizeIndex = 2; contentIndex = 3; } else { statement = insertStatement; segmentIdIndex = 4; attachmentSizeIndex = 5; contentIndex = 6; } statement.setInt(segmentIdIndex, currentSegmentId); if (attachment.getContent().length <= chunkSize) { statement.setInt(attachmentSizeIndex, attachment.getContent().length); statement.setBytes(contentIndex, attachment.getContent()); statement.executeUpdate(); currentSegmentId++; done = true; } else { if (inputStream == null) { inputStream = new ByteArrayInputStream(attachment.getContent()); } if (inputStream.available() > 0) { int segmentSize = Math.min(chunkSize, inputStream.available()); byte[] segment = new byte[segmentSize]; inputStream.read(segment, 0, segmentSize); statement.setInt(attachmentSizeIndex, segmentSize); statement.setBytes(contentIndex, segment); statement.executeUpdate(); currentSegmentId++; } else { done = true; } } } updateStatement.clearParameters(); insertStatement.clearParameters(); if (totalSegmentCount >= currentSegmentId) { PreparedStatement deleteStatement = prepareStatement("deleteMessageAttachmentLingeringSegments", channelId); deleteStatement.setString(1, attachment.getId()); deleteStatement.setLong(2, messageId); deleteStatement.setInt(3, currentSegmentId); deleteStatement.executeUpdate(); } } catch (SQLException e) { throw new DonkeyDaoException(e); } finally { List<AutoCloseable> dbObjects = new ArrayList<>(); dbObjects.add(segmentCountStatement); dbObjects.add(segmentCountResult); dbObjects.add(updateStatement); dbObjects.add(insertStatement); closeDatabaseObjectsIfNeeded(dbObjects); } } }
JdbcDao implements DonkeyDao { @Override public void updateMessageAttachment(String channelId, long messageId, Attachment attachment) { logger.debug(channelId + "/" + messageId + ": updating message attachment"); PreparedStatement segmentCountStatement = null; ResultSet segmentCountResult = null; PreparedStatement updateStatement = null; PreparedStatement insertStatement = null; try { segmentCountStatement = prepareStatement("selectMessageAttachmentSegmentCount", channelId); segmentCountStatement.setString(1, attachment.getId()); segmentCountStatement.setLong(2, messageId); segmentCountResult = segmentCountStatement.executeQuery(); segmentCountResult.next(); int totalSegmentCount = segmentCountResult.getInt(1); updateStatement = prepareStatement("updateMessageAttachment", channelId); updateStatement.setString(1, attachment.getType()); updateStatement.setString(4, attachment.getId()); updateStatement.setLong(5, messageId); insertStatement = prepareStatement("insertMessageAttachment", channelId); insertStatement.setString(1, attachment.getId()); insertStatement.setLong(2, messageId); insertStatement.setString(3, attachment.getType()); int chunkSize = 10000000; ByteArrayInputStream inputStream = null; boolean done = false; int currentSegmentId = 1; while (!done) { PreparedStatement statement; int segmentIdIndex; int attachmentSizeIndex; int contentIndex; if (currentSegmentId <= totalSegmentCount) { statement = updateStatement; segmentIdIndex = 6; attachmentSizeIndex = 2; contentIndex = 3; } else { statement = insertStatement; segmentIdIndex = 4; attachmentSizeIndex = 5; contentIndex = 6; } statement.setInt(segmentIdIndex, currentSegmentId); if (attachment.getContent().length <= chunkSize) { statement.setInt(attachmentSizeIndex, attachment.getContent().length); statement.setBytes(contentIndex, attachment.getContent()); statement.executeUpdate(); currentSegmentId++; done = true; } else { if (inputStream == null) { inputStream = new ByteArrayInputStream(attachment.getContent()); } if (inputStream.available() > 0) { int segmentSize = Math.min(chunkSize, inputStream.available()); byte[] segment = new byte[segmentSize]; inputStream.read(segment, 0, segmentSize); statement.setInt(attachmentSizeIndex, segmentSize); statement.setBytes(contentIndex, segment); statement.executeUpdate(); currentSegmentId++; } else { done = true; } } } updateStatement.clearParameters(); insertStatement.clearParameters(); if (totalSegmentCount >= currentSegmentId) { PreparedStatement deleteStatement = prepareStatement("deleteMessageAttachmentLingeringSegments", channelId); deleteStatement.setString(1, attachment.getId()); deleteStatement.setLong(2, messageId); deleteStatement.setInt(3, currentSegmentId); deleteStatement.executeUpdate(); } } catch (SQLException e) { throw new DonkeyDaoException(e); } finally { List<AutoCloseable> dbObjects = new ArrayList<>(); dbObjects.add(segmentCountStatement); dbObjects.add(segmentCountResult); dbObjects.add(updateStatement); dbObjects.add(insertStatement); closeDatabaseObjectsIfNeeded(dbObjects); } } protected  JdbcDao(Donkey donkey, Connection connection, QuerySource querySource, PreparedStatementSource statementSource, SerializerProvider serializerProvider, boolean encryptData, boolean decryptData, StatisticsUpdater statisticsUpdater, Statistics currentStats, Statistics totalStats, String statsServerId); }
JdbcDao implements DonkeyDao { @Override public void updateMessageAttachment(String channelId, long messageId, Attachment attachment) { logger.debug(channelId + "/" + messageId + ": updating message attachment"); PreparedStatement segmentCountStatement = null; ResultSet segmentCountResult = null; PreparedStatement updateStatement = null; PreparedStatement insertStatement = null; try { segmentCountStatement = prepareStatement("selectMessageAttachmentSegmentCount", channelId); segmentCountStatement.setString(1, attachment.getId()); segmentCountStatement.setLong(2, messageId); segmentCountResult = segmentCountStatement.executeQuery(); segmentCountResult.next(); int totalSegmentCount = segmentCountResult.getInt(1); updateStatement = prepareStatement("updateMessageAttachment", channelId); updateStatement.setString(1, attachment.getType()); updateStatement.setString(4, attachment.getId()); updateStatement.setLong(5, messageId); insertStatement = prepareStatement("insertMessageAttachment", channelId); insertStatement.setString(1, attachment.getId()); insertStatement.setLong(2, messageId); insertStatement.setString(3, attachment.getType()); int chunkSize = 10000000; ByteArrayInputStream inputStream = null; boolean done = false; int currentSegmentId = 1; while (!done) { PreparedStatement statement; int segmentIdIndex; int attachmentSizeIndex; int contentIndex; if (currentSegmentId <= totalSegmentCount) { statement = updateStatement; segmentIdIndex = 6; attachmentSizeIndex = 2; contentIndex = 3; } else { statement = insertStatement; segmentIdIndex = 4; attachmentSizeIndex = 5; contentIndex = 6; } statement.setInt(segmentIdIndex, currentSegmentId); if (attachment.getContent().length <= chunkSize) { statement.setInt(attachmentSizeIndex, attachment.getContent().length); statement.setBytes(contentIndex, attachment.getContent()); statement.executeUpdate(); currentSegmentId++; done = true; } else { if (inputStream == null) { inputStream = new ByteArrayInputStream(attachment.getContent()); } if (inputStream.available() > 0) { int segmentSize = Math.min(chunkSize, inputStream.available()); byte[] segment = new byte[segmentSize]; inputStream.read(segment, 0, segmentSize); statement.setInt(attachmentSizeIndex, segmentSize); statement.setBytes(contentIndex, segment); statement.executeUpdate(); currentSegmentId++; } else { done = true; } } } updateStatement.clearParameters(); insertStatement.clearParameters(); if (totalSegmentCount >= currentSegmentId) { PreparedStatement deleteStatement = prepareStatement("deleteMessageAttachmentLingeringSegments", channelId); deleteStatement.setString(1, attachment.getId()); deleteStatement.setLong(2, messageId); deleteStatement.setInt(3, currentSegmentId); deleteStatement.executeUpdate(); } } catch (SQLException e) { throw new DonkeyDaoException(e); } finally { List<AutoCloseable> dbObjects = new ArrayList<>(); dbObjects.add(segmentCountStatement); dbObjects.add(segmentCountResult); dbObjects.add(updateStatement); dbObjects.add(insertStatement); closeDatabaseObjectsIfNeeded(dbObjects); } } protected  JdbcDao(Donkey donkey, Connection connection, QuerySource querySource, PreparedStatementSource statementSource, SerializerProvider serializerProvider, boolean encryptData, boolean decryptData, StatisticsUpdater statisticsUpdater, Statistics currentStats, Statistics totalStats, String statsServerId); @Override void setEncryptData(boolean encryptData); @Override void setDecryptData(boolean decryptData); @Override void setStatisticsUpdater(StatisticsUpdater statisticsUpdater); char getQuoteChar(); void setQuoteChar(char quoteChar); @Override void insertMessage(Message message); @Override void updateSendAttempts(ConnectorMessage connectorMessage); @Override void insertMessageContent(MessageContent messageContent); @Override void batchInsertMessageContent(MessageContent messageContent); @Override void executeBatchInsertMessageContent(String channelId); @Override void storeMessageContent(MessageContent messageContent); void storeContent(String channelId, long messageId, int metaDataId, ContentType contentType, String content, String dataType, boolean encrypted); @Override void addChannelStatistics(Statistics statistics); @Override void insertMessageAttachment(String channelId, long messageId, Attachment attachment); @Override void updateMessageAttachment(String channelId, long messageId, Attachment attachment); @Override void insertMetaData(ConnectorMessage connectorMessage, List<MetaDataColumn> metaDataColumns); @Override void storeMetaData(ConnectorMessage connectorMessage, List<MetaDataColumn> metaDataColumns); @Override void insertConnectorMessage(ConnectorMessage connectorMessage, boolean storeMaps, boolean updateStats); @Override void deleteMessage(String channelId, long messageId); @Override void deleteConnectorMessages(String channelId, long messageId, Set<Integer> metaDataIds); @Override void deleteMessageStatistics(String channelId, long messageId, Set<Integer> metaDataIds); @Override void updateStatus(ConnectorMessage connectorMessage, Status previousStatus); @Override void updateErrors(ConnectorMessage connectorMessage); @Override void updateMaps(ConnectorMessage connectorMessage); @Override void updateSourceMap(ConnectorMessage connectorMessage); @Override void updateResponseMap(ConnectorMessage connectorMessage); @Override void markAsProcessed(String channelId, long messageId); @Override void resetMessage(String channelId, long messageId); @Override Map<String, Long> getLocalChannelIds(); @Override void removeChannel(String channelId); @Override Long selectMaxLocalChannelId(); @Override void deleteAllMessages(String channelId); @Override void deleteMessageContent(String channelId, long messageId); @Override void deleteMessageContentByMetaDataIds(String channelId, long messageId, Set<Integer> metaDataIds); @Override void deleteMessageAttachments(String channelId, long messageId); @Override List<MetaDataColumn> getMetaDataColumns(String channelId); @Override void removeMetaDataColumn(String channelId, String columnName); @Override long getMaxMessageId(String channelId); @Override long getMinMessageId(String channelId); @Override long getNextMessageId(String channelId); @Override List<Attachment> getMessageAttachment(String channelId, long messageId); @Override Attachment getMessageAttachment(String channelId, String attachmentId, Long messageId); @Override List<Message> getUnfinishedMessages(String channelId, String serverId, int limit, Long minMessageId); @Override List<Message> getPendingConnectorMessages(String channelId, String serverId, int limit, Long minMessageId); @Override List<Message> getMessages(String channelId, List<Long> messageIds); @Override List<ConnectorMessage> getConnectorMessages(String channelId, String serverId, int metaDataId, Status status, int offset, int limit, Long minMessageId, Long maxMessageId); @Override List<ConnectorMessage> getConnectorMessages(String channelId, long messageId, Set<Integer> metaDataIds, boolean includeContent); @Override Map<Integer, ConnectorMessage> getConnectorMessages(String channelId, long messageId, List<Integer> metaDataIds); @Override Map<Integer, Status> getConnectorMessageStatuses(String channelId, long messageId, boolean checkProcessed); @Override int getConnectorMessageCount(String channelId, String serverId, int metaDataId, Status status); @Override long getConnectorMessageMaxMessageId(String channelId, String serverId, int metaDataId, Status status); @Override void addMetaDataColumn(String channelId, MetaDataColumn metaDataColumn); @Override void createChannel(String channelId, long localChannelId); @Override void checkAndCreateChannelTables(); @Override void resetStatistics(String channelId, Integer metaDataId, Set<Status> statuses); @Override void resetAllStatistics(String channelId); @Override Statistics getChannelStatistics(String serverId); @Override Statistics getChannelTotalStatistics(String serverId); @Override void commit(); @Override void commit(boolean durable); @Override void rollback(); @Override void close(); @Override boolean isClosed(); @Override boolean initTableStructure(); Connection getConnection(); String getAsyncCommitCommand(); void setAsyncCommitCommand(String asyncCommitCommand); }
JdbcDao implements DonkeyDao { @Override public void updateMessageAttachment(String channelId, long messageId, Attachment attachment) { logger.debug(channelId + "/" + messageId + ": updating message attachment"); PreparedStatement segmentCountStatement = null; ResultSet segmentCountResult = null; PreparedStatement updateStatement = null; PreparedStatement insertStatement = null; try { segmentCountStatement = prepareStatement("selectMessageAttachmentSegmentCount", channelId); segmentCountStatement.setString(1, attachment.getId()); segmentCountStatement.setLong(2, messageId); segmentCountResult = segmentCountStatement.executeQuery(); segmentCountResult.next(); int totalSegmentCount = segmentCountResult.getInt(1); updateStatement = prepareStatement("updateMessageAttachment", channelId); updateStatement.setString(1, attachment.getType()); updateStatement.setString(4, attachment.getId()); updateStatement.setLong(5, messageId); insertStatement = prepareStatement("insertMessageAttachment", channelId); insertStatement.setString(1, attachment.getId()); insertStatement.setLong(2, messageId); insertStatement.setString(3, attachment.getType()); int chunkSize = 10000000; ByteArrayInputStream inputStream = null; boolean done = false; int currentSegmentId = 1; while (!done) { PreparedStatement statement; int segmentIdIndex; int attachmentSizeIndex; int contentIndex; if (currentSegmentId <= totalSegmentCount) { statement = updateStatement; segmentIdIndex = 6; attachmentSizeIndex = 2; contentIndex = 3; } else { statement = insertStatement; segmentIdIndex = 4; attachmentSizeIndex = 5; contentIndex = 6; } statement.setInt(segmentIdIndex, currentSegmentId); if (attachment.getContent().length <= chunkSize) { statement.setInt(attachmentSizeIndex, attachment.getContent().length); statement.setBytes(contentIndex, attachment.getContent()); statement.executeUpdate(); currentSegmentId++; done = true; } else { if (inputStream == null) { inputStream = new ByteArrayInputStream(attachment.getContent()); } if (inputStream.available() > 0) { int segmentSize = Math.min(chunkSize, inputStream.available()); byte[] segment = new byte[segmentSize]; inputStream.read(segment, 0, segmentSize); statement.setInt(attachmentSizeIndex, segmentSize); statement.setBytes(contentIndex, segment); statement.executeUpdate(); currentSegmentId++; } else { done = true; } } } updateStatement.clearParameters(); insertStatement.clearParameters(); if (totalSegmentCount >= currentSegmentId) { PreparedStatement deleteStatement = prepareStatement("deleteMessageAttachmentLingeringSegments", channelId); deleteStatement.setString(1, attachment.getId()); deleteStatement.setLong(2, messageId); deleteStatement.setInt(3, currentSegmentId); deleteStatement.executeUpdate(); } } catch (SQLException e) { throw new DonkeyDaoException(e); } finally { List<AutoCloseable> dbObjects = new ArrayList<>(); dbObjects.add(segmentCountStatement); dbObjects.add(segmentCountResult); dbObjects.add(updateStatement); dbObjects.add(insertStatement); closeDatabaseObjectsIfNeeded(dbObjects); } } protected  JdbcDao(Donkey donkey, Connection connection, QuerySource querySource, PreparedStatementSource statementSource, SerializerProvider serializerProvider, boolean encryptData, boolean decryptData, StatisticsUpdater statisticsUpdater, Statistics currentStats, Statistics totalStats, String statsServerId); @Override void setEncryptData(boolean encryptData); @Override void setDecryptData(boolean decryptData); @Override void setStatisticsUpdater(StatisticsUpdater statisticsUpdater); char getQuoteChar(); void setQuoteChar(char quoteChar); @Override void insertMessage(Message message); @Override void updateSendAttempts(ConnectorMessage connectorMessage); @Override void insertMessageContent(MessageContent messageContent); @Override void batchInsertMessageContent(MessageContent messageContent); @Override void executeBatchInsertMessageContent(String channelId); @Override void storeMessageContent(MessageContent messageContent); void storeContent(String channelId, long messageId, int metaDataId, ContentType contentType, String content, String dataType, boolean encrypted); @Override void addChannelStatistics(Statistics statistics); @Override void insertMessageAttachment(String channelId, long messageId, Attachment attachment); @Override void updateMessageAttachment(String channelId, long messageId, Attachment attachment); @Override void insertMetaData(ConnectorMessage connectorMessage, List<MetaDataColumn> metaDataColumns); @Override void storeMetaData(ConnectorMessage connectorMessage, List<MetaDataColumn> metaDataColumns); @Override void insertConnectorMessage(ConnectorMessage connectorMessage, boolean storeMaps, boolean updateStats); @Override void deleteMessage(String channelId, long messageId); @Override void deleteConnectorMessages(String channelId, long messageId, Set<Integer> metaDataIds); @Override void deleteMessageStatistics(String channelId, long messageId, Set<Integer> metaDataIds); @Override void updateStatus(ConnectorMessage connectorMessage, Status previousStatus); @Override void updateErrors(ConnectorMessage connectorMessage); @Override void updateMaps(ConnectorMessage connectorMessage); @Override void updateSourceMap(ConnectorMessage connectorMessage); @Override void updateResponseMap(ConnectorMessage connectorMessage); @Override void markAsProcessed(String channelId, long messageId); @Override void resetMessage(String channelId, long messageId); @Override Map<String, Long> getLocalChannelIds(); @Override void removeChannel(String channelId); @Override Long selectMaxLocalChannelId(); @Override void deleteAllMessages(String channelId); @Override void deleteMessageContent(String channelId, long messageId); @Override void deleteMessageContentByMetaDataIds(String channelId, long messageId, Set<Integer> metaDataIds); @Override void deleteMessageAttachments(String channelId, long messageId); @Override List<MetaDataColumn> getMetaDataColumns(String channelId); @Override void removeMetaDataColumn(String channelId, String columnName); @Override long getMaxMessageId(String channelId); @Override long getMinMessageId(String channelId); @Override long getNextMessageId(String channelId); @Override List<Attachment> getMessageAttachment(String channelId, long messageId); @Override Attachment getMessageAttachment(String channelId, String attachmentId, Long messageId); @Override List<Message> getUnfinishedMessages(String channelId, String serverId, int limit, Long minMessageId); @Override List<Message> getPendingConnectorMessages(String channelId, String serverId, int limit, Long minMessageId); @Override List<Message> getMessages(String channelId, List<Long> messageIds); @Override List<ConnectorMessage> getConnectorMessages(String channelId, String serverId, int metaDataId, Status status, int offset, int limit, Long minMessageId, Long maxMessageId); @Override List<ConnectorMessage> getConnectorMessages(String channelId, long messageId, Set<Integer> metaDataIds, boolean includeContent); @Override Map<Integer, ConnectorMessage> getConnectorMessages(String channelId, long messageId, List<Integer> metaDataIds); @Override Map<Integer, Status> getConnectorMessageStatuses(String channelId, long messageId, boolean checkProcessed); @Override int getConnectorMessageCount(String channelId, String serverId, int metaDataId, Status status); @Override long getConnectorMessageMaxMessageId(String channelId, String serverId, int metaDataId, Status status); @Override void addMetaDataColumn(String channelId, MetaDataColumn metaDataColumn); @Override void createChannel(String channelId, long localChannelId); @Override void checkAndCreateChannelTables(); @Override void resetStatistics(String channelId, Integer metaDataId, Set<Status> statuses); @Override void resetAllStatistics(String channelId); @Override Statistics getChannelStatistics(String serverId); @Override Statistics getChannelTotalStatistics(String serverId); @Override void commit(); @Override void commit(boolean durable); @Override void rollback(); @Override void close(); @Override boolean isClosed(); @Override boolean initTableStructure(); Connection getConnection(); String getAsyncCommitCommand(); void setAsyncCommitCommand(String asyncCommitCommand);  }