Package org.jumpmind.symmetric.service.impl

Class DataExtractorService

java.lang.Object

org.jumpmind.symmetric.service.impl.AbstractService

org.jumpmind.symmetric.service.impl.DataExtractorService

All Implemented Interfaces:

IDataExtractorService, INodeCommunicationService.INodeCommunicationExecutor, IService

Direct Known Subclasses:

FileSyncExtractorService

public class DataExtractorService
extends AbstractService
implements IDataExtractorService, INodeCommunicationService.INodeCommunicationExecutor

See Also:

• IDataExtractorService

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
protected static enum	DataExtractorService.ExtractMode

Field Summary

Fields

Modifier and Type	Field	Description
protected ISymmetricEngine	engine

Fields inherited from class org.jumpmind.symmetric.service.impl.AbstractService

log, parameterService, platform, sqlTemplate, sqlTemplateDirty, symmetricDialect, tablePrefix

Constructor Summary

Constructors

Constructor	Description
DataExtractorService(ISymmetricEngine engine)

Method Summary

Modifier and Type	Method	Description
StagingFileLock	acquireStagingFileLock(OutgoingBatch batch)
protected ExtractDataReader	buildExtractDataReader(Node sourceNode, Node targetNode, OutgoingBatch currentBatch, ProcessInfo processInfo)
protected ExtractDataReader	buildExtractDataReader(Node sourceNode, Node targetNode, OutgoingBatch currentBatch, ProcessInfo processInfo, boolean containsBigLob)
protected MultiBatchStagingWriter	buildMultiBatchStagingWriter(ExtractRequest request, List<ExtractRequest> childRequests, Node sourceNode, Node targetNode, List<OutgoingBatch> batches, ProcessInfo processInfo, Channel channel, boolean isRestarted)
int	cancelExtractRequests(long loadId)
protected boolean	canProcessExtractRequest(ExtractRequest request, NodeCommunication.CommunicationType communicationType)
protected final boolean	changeBatchStatus(AbstractBatch.Status status, OutgoingBatch currentBatch, DataExtractorService.ExtractMode mode)
protected void	checkSendDeferredConstraints(ExtractRequest request, List<ExtractRequest> childRequests, Node targetNode)
protected void	cleanupIgnoredBatch(Node sourceNode, Node targetNode, OutgoingBatch currentBatch, IDataWriter writer)
protected TransformWriter	createTransformDataWriter(Node identity, Node targetNode, IDataWriter extractWriter)
void	execute(NodeCommunication nodeCommunication, RemoteNodeStatus status)	This is a callback method used by the NodeCommunicationService that extracts an initial load in the background.
List<OutgoingBatch>	extract(ProcessInfo extractInfo, Node targetNode, String queue, IOutgoingTransport transport)
protected List<OutgoingBatch>	extract(ProcessInfo extractInfo, Node targetNode, List<OutgoingBatch> activeBatches, IDataWriter dataWriter, BufferedWriter writer, DataExtractorService.ExtractMode mode)
List<OutgoingBatch>	extract(ProcessInfo extractInfo, Node targetNode, IOutgoingTransport transport)
protected org.jumpmind.symmetric.service.impl.DataExtractorService.FutureOutgoingBatch	extractBatch(OutgoingBatch extractBatch, org.jumpmind.symmetric.service.impl.DataExtractorService.FutureExtractStatus status, ProcessInfo extractInfo, Node targetNode, IDataWriter dataWriter, DataExtractorService.ExtractMode mode, List<OutgoingBatch> activeBatches)
boolean	extractBatchRange(Writer writer, String nodeId, long startBatchId, long endBatchId)
boolean	extractBatchRange(Writer writer, String nodeId, Date startBatchTime, Date endBatchTime, String... channelIds)
void	extractConfigurationStandalone(Node targetNode, Writer writer, String... tablesToExclude)	Extract the SymmetricDS configuration for the passed in Node.
boolean	extractOnlyOutgoingBatch(String nodeId, long batchId, Writer writer)	This method will extract an outgoing batch, but will not update the outgoing batch status
protected OutgoingBatch	extractOutgoingBatch(ProcessInfo extractInfo, Node targetNode, IDataWriter dataWriter, OutgoingBatch currentBatch, boolean useStagingDataWriter, boolean updateBatchStatistics, DataExtractorService.ExtractMode mode, IDataProcessorListener listener)
List<OutgoingBatchWithPayload>	extractToPayload(ProcessInfo processInfo, Node targetNode, StructureDataWriter.PayloadType payloadType, boolean useJdbcTimestampFormat, boolean useUpsertStatements, boolean useDelimiterIdentifiers)
protected int	findStatsIndex(String bufferString, String prevBuffer)
protected String	getBatchStats(OutgoingBatch batch)
protected String	getBatchStatsColumns()
List<ExtractRequest>	getCompletedTablesForExtractByLoadId(long loadId)
List<ExtractRequest>	getCompletedTablesForExtractByLoadIdAndNodeId(long loadId, String nodeId)
protected List<ExtractRequest>	getExtractChildRequestsForNode(ExtractRequest parentRequest)
protected Map<Long,List<ExtractRequest>>	getExtractChildRequestsForNode(NodeCommunication nodeCommunication, String queue, QueueThread queueThread)
protected ExtractRequest	getExtractRequestForBatch(OutgoingBatch batch)
List<org.jumpmind.symmetric.service.impl.DataExtractorService.NodeQueuePair>	getExtractRequestNodes()
protected List<ExtractRequest>	getExtractRequestsForNode(NodeCommunication nodeCommunication, String queue, QueueThread queueThread)
protected List<ExtractRequest>	getExtractRequestsForThreading()
protected Statistics	getExtractStats(IDataWriter writer, OutgoingBatch currentBatch)
List<ExtractRequest>	getPendingTablesForExtractByLoadId(long loadId)
List<ExtractRequest>	getPendingTablesForExtractByLoadIdAndNodeId(long loadId, String nodeId)
protected ProcessType	getProcessType()
protected String	getSemaphoreKey(OutgoingBatch batch, boolean useStagingDataWriter)
protected IStagedResource	getStagedResource(OutgoingBatch currentBatch)
protected boolean	isApplicable(NodeCommunication nodeCommunication)
protected boolean	isPreviouslyExtracted(OutgoingBatch currentBatch, boolean acquireReference)
protected boolean	isRetry(OutgoingBatch currentBatch, Node remoteNode)
protected boolean	isStagingFileLockRequired(OutgoingBatch batch)
protected OutgoingBatches	loadPendingBatches(ProcessInfo extractInfo, Node targetNode, String queue, IOutgoingTransport transport)
protected void	queue(String nodeId, String queue, RemoteNodeStatuses statuses)
RemoteNodeStatuses	queueWork(boolean force)
protected void	releaseLock(org.jumpmind.symmetric.service.impl.DataExtractorService.BatchLock lock, OutgoingBatch batch, boolean useStagingDataWriter)
void	releaseMissedExtractRequests()
void	removeBatchFromStaging(OutgoingBatch batch)
protected final OutgoingBatch	requeryIfEnoughTimeHasPassed(long ts, OutgoingBatch currentBatch)	If time has passed, then re-query the batch to double check that the status has not changed
ExtractRequest	requestExtractRequest(ISqlTransaction transaction, String nodeId, String queue, TriggerRouter triggerRouter, long startBatchId, long endBatchId, long loadId, String table, long rows, long parentRequestId)
void	resetExtractRequest(OutgoingBatch batch)
protected void	restartExtractRequest(List<OutgoingBatch> batches, ExtractRequest request, List<ExtractRequest> childRequests)
protected OutgoingBatch	sendOutgoingBatch(ProcessInfo processInfo, Node targetNode, OutgoingBatch currentBatch, boolean isRetry, IDataWriter dataWriter, BufferedWriter writer, DataExtractorService.ExtractMode mode)
protected void	transferFromStaging(DataExtractorService.ExtractMode mode, Batch.BatchType batchType, OutgoingBatch batch, boolean isRetry, IStagedResource stagedResource, BufferedWriter writer, DataContext context, BigDecimal maxKBytesPerSec, ProcessInfo processInfo)
protected void	triggerReExtraction(OutgoingBatch currentBatch)
void	updateExtractRequestLoadTime(ISqlTransaction transaction, Date loadTime, OutgoingBatch outgoingBatch)
protected void	updateExtractRequestsForThreading()
protected void	updateExtractRequestsForThreading(Collection<ExtractRequest> extractRequests)
protected void	updateExtractRequestStatus(ISqlTransaction transaction, long extractId, ExtractRequest.ExtractStatus status, long extractedRows, long extractedMillis)
void	updateExtractRequestStatuses(ISqlTransaction transaction, long loadId, String sourceNodeId, String fromStatus, String toStatus)
void	updateExtractRequestTransferred(OutgoingBatch batch, long transferMillis)
protected IDataWriter	wrapWithTransformWriter(Node sourceNode, Node targetNode, ProcessInfo processInfo, IDataWriter dataWriter, boolean useStagingDataWriter)
protected boolean	writeBatchStats(BufferedWriter writer, char[] buffer, int bufferSize, String prevBuffer, OutgoingBatch batch)
protected void	writeKeepAliveAck(BufferedWriter writer, Node sourceNode, boolean streamToFileEnabled)

Methods inherited from class org.jumpmind.symmetric.service.impl.AbstractService

assertNotNull, buildBatchOrderBy, buildBatchParams, buildBatchWhere, buildBatchWhereFromFilter, close, createSqlReplacementTokens, getJdbcTemplate, getParameterService, getSql, getSymmetricDialect, getTablePrefix, getTargetDialect, getTargetPlatform, getTargetPlatform, isCalledFromSymmetricAdminTool, isSet, isStreamClosedByClient, isSymmetricTable, logOnce, maxDate, readAcks, sendAck, setSqlMap, synchronize, toNodeIds, toNodeIds

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Details

engine

protected ISymmetricEngine engine

Constructor Details

DataExtractorService

public DataExtractorService(ISymmetricEngine engine)

Method Details

extractConfigurationStandalone

public void extractConfigurationStandalone(Node targetNode,
 Writer writer,
 String... tablesToExclude)

Extract the SymmetricDS configuration for the passed in Node.

Specified by:

extractConfigurationStandalone in interface IDataExtractorService

extractToPayload

public List<OutgoingBatchWithPayload> extractToPayload(ProcessInfo processInfo,
 Node targetNode,
 StructureDataWriter.PayloadType payloadType,
 boolean useJdbcTimestampFormat,
 boolean useUpsertStatements,
 boolean useDelimiterIdentifiers)

Specified by:

extractToPayload in interface IDataExtractorService

extract

public List<OutgoingBatch> extract(ProcessInfo extractInfo,
 Node targetNode,
 IOutgoingTransport transport)

Specified by:

extract in interface IDataExtractorService

Returns:

a list of batches that were extracted

extract

public List<OutgoingBatch> extract(ProcessInfo extractInfo,
 Node targetNode,
 String queue,
 IOutgoingTransport transport)

Specified by:

extract in interface IDataExtractorService

loadPendingBatches

protected OutgoingBatches loadPendingBatches(ProcessInfo extractInfo,
 Node targetNode,
 String queue,
 IOutgoingTransport transport)

extractOnlyOutgoingBatch

public boolean extractOnlyOutgoingBatch(String nodeId,
 long batchId,
 Writer writer)

This method will extract an outgoing batch, but will not update the outgoing batch status

Specified by:

extractOnlyOutgoingBatch in interface IDataExtractorService

extract

protected List<OutgoingBatch> extract(ProcessInfo extractInfo,
 Node targetNode,
 List<OutgoingBatch> activeBatches,
 IDataWriter dataWriter,
 BufferedWriter writer,
 DataExtractorService.ExtractMode mode)

extractBatch

protected org.jumpmind.symmetric.service.impl.DataExtractorService.FutureOutgoingBatch extractBatch(OutgoingBatch extractBatch,
 org.jumpmind.symmetric.service.impl.DataExtractorService.FutureExtractStatus status,
 ProcessInfo extractInfo,
 Node targetNode,
 IDataWriter dataWriter,
 DataExtractorService.ExtractMode mode,
 List<OutgoingBatch> activeBatches)
                                                                                             throws Exception

Throws:

Exception

writeKeepAliveAck

protected void writeKeepAliveAck(BufferedWriter writer,
 Node sourceNode,
 boolean streamToFileEnabled)

changeBatchStatus

protected final boolean changeBatchStatus(AbstractBatch.Status status,
 OutgoingBatch currentBatch,
 DataExtractorService.ExtractMode mode)

requeryIfEnoughTimeHasPassed

protected final OutgoingBatch requeryIfEnoughTimeHasPassed(long ts,
 OutgoingBatch currentBatch)

If time has passed, then re-query the batch to double check that the status has not changed

extractOutgoingBatch

protected OutgoingBatch extractOutgoingBatch(ProcessInfo extractInfo,
 Node targetNode,
 IDataWriter dataWriter,
 OutgoingBatch currentBatch,
 boolean useStagingDataWriter,
 boolean updateBatchStatistics,
 DataExtractorService.ExtractMode mode,
 IDataProcessorListener listener)

getSemaphoreKey

protected String getSemaphoreKey(OutgoingBatch batch,
 boolean useStagingDataWriter)

acquireStagingFileLock

public StagingFileLock acquireStagingFileLock(OutgoingBatch batch)

Specified by:

acquireStagingFileLock in interface IDataExtractorService

releaseLock

protected void releaseLock(org.jumpmind.symmetric.service.impl.DataExtractorService.BatchLock lock,
 OutgoingBatch batch,
 boolean useStagingDataWriter)

isStagingFileLockRequired

protected boolean isStagingFileLockRequired(OutgoingBatch batch)

triggerReExtraction

protected void triggerReExtraction(OutgoingBatch currentBatch)

buildExtractDataReader

protected ExtractDataReader buildExtractDataReader(Node sourceNode,
 Node targetNode,
 OutgoingBatch currentBatch,
 ProcessInfo processInfo)

buildExtractDataReader

protected ExtractDataReader buildExtractDataReader(Node sourceNode,
 Node targetNode,
 OutgoingBatch currentBatch,
 ProcessInfo processInfo,
 boolean containsBigLob)

getExtractStats

protected Statistics getExtractStats(IDataWriter writer,
 OutgoingBatch currentBatch)

wrapWithTransformWriter

protected IDataWriter wrapWithTransformWriter(Node sourceNode,
 Node targetNode,
 ProcessInfo processInfo,
 IDataWriter dataWriter,
 boolean useStagingDataWriter)

cleanupIgnoredBatch

protected void cleanupIgnoredBatch(Node sourceNode,
 Node targetNode,
 OutgoingBatch currentBatch,
 IDataWriter writer)

getStagedResource

protected IStagedResource getStagedResource(OutgoingBatch currentBatch)

isPreviouslyExtracted

protected boolean isPreviouslyExtracted(OutgoingBatch currentBatch,
 boolean acquireReference)

isRetry

protected boolean isRetry(OutgoingBatch currentBatch,
 Node remoteNode)

sendOutgoingBatch

protected OutgoingBatch sendOutgoingBatch(ProcessInfo processInfo,
 Node targetNode,
 OutgoingBatch currentBatch,
 boolean isRetry,
 IDataWriter dataWriter,
 BufferedWriter writer,
 DataExtractorService.ExtractMode mode)

transferFromStaging

protected void transferFromStaging(DataExtractorService.ExtractMode mode,
 Batch.BatchType batchType,
 OutgoingBatch batch,
 boolean isRetry,
 IStagedResource stagedResource,
 BufferedWriter writer,
 DataContext context,
 BigDecimal maxKBytesPerSec,
 ProcessInfo processInfo)

findStatsIndex

protected int findStatsIndex(String bufferString,
 String prevBuffer)

getPendingTablesForExtractByLoadId

public List<ExtractRequest> getPendingTablesForExtractByLoadId(long loadId)

Specified by:

getPendingTablesForExtractByLoadId in interface IDataExtractorService

getCompletedTablesForExtractByLoadId

public List<ExtractRequest> getCompletedTablesForExtractByLoadId(long loadId)

Specified by:

getCompletedTablesForExtractByLoadId in interface IDataExtractorService

getPendingTablesForExtractByLoadIdAndNodeId

public List<ExtractRequest> getPendingTablesForExtractByLoadIdAndNodeId(long loadId,
 String nodeId)

Specified by:

getPendingTablesForExtractByLoadIdAndNodeId in interface IDataExtractorService

getCompletedTablesForExtractByLoadIdAndNodeId

public List<ExtractRequest> getCompletedTablesForExtractByLoadIdAndNodeId(long loadId,
 String nodeId)

Specified by:

getCompletedTablesForExtractByLoadIdAndNodeId in interface IDataExtractorService

updateExtractRequestLoadTime

public void updateExtractRequestLoadTime(ISqlTransaction transaction,
 Date loadTime,
 OutgoingBatch outgoingBatch)

Specified by:

updateExtractRequestLoadTime in interface IDataExtractorService

updateExtractRequestTransferred

public void updateExtractRequestTransferred(OutgoingBatch batch,
 long transferMillis)

Specified by:

updateExtractRequestTransferred in interface IDataExtractorService

cancelExtractRequests

public int cancelExtractRequests(long loadId)

Specified by:

cancelExtractRequests in interface IDataExtractorService

writeBatchStats

protected boolean writeBatchStats(BufferedWriter writer,
 char[] buffer,
 int bufferSize,
 String prevBuffer,
 OutgoingBatch batch)
                           throws IOException

Throws:

IOException

getBatchStatsColumns

protected String getBatchStatsColumns()

getBatchStats

protected String getBatchStats(OutgoingBatch batch)

extractBatchRange

public boolean extractBatchRange(Writer writer,
 String nodeId,
 long startBatchId,
 long endBatchId)

Specified by:

extractBatchRange in interface IDataExtractorService

extractBatchRange

public boolean extractBatchRange(Writer writer,
 String nodeId,
 Date startBatchTime,
 Date endBatchTime,
 String... channelIds)

Specified by:

extractBatchRange in interface IDataExtractorService

createTransformDataWriter

protected TransformWriter createTransformDataWriter(Node identity,
 Node targetNode,
 IDataWriter extractWriter)

queueWork

public RemoteNodeStatuses queueWork(boolean force)

Specified by:

queueWork in interface IDataExtractorService

queue

protected void queue(String nodeId,
 String queue,
 RemoteNodeStatuses statuses)

getExtractRequestNodes

public List<org.jumpmind.symmetric.service.impl.DataExtractorService.NodeQueuePair> getExtractRequestNodes()

updateExtractRequestsForThreading

protected void updateExtractRequestsForThreading()

getExtractRequestsForThreading

protected List<ExtractRequest> getExtractRequestsForThreading()

updateExtractRequestsForThreading

protected void updateExtractRequestsForThreading(Collection<ExtractRequest> extractRequests)

getExtractRequestsForNode

protected List<ExtractRequest> getExtractRequestsForNode(NodeCommunication nodeCommunication,
 String queue,
 QueueThread queueThread)

getExtractRequestForBatch

protected ExtractRequest getExtractRequestForBatch(OutgoingBatch batch)

getExtractChildRequestsForNode

protected Map<Long,List<ExtractRequest>> getExtractChildRequestsForNode(NodeCommunication nodeCommunication,
 String queue,
 QueueThread queueThread)

getExtractChildRequestsForNode

protected List<ExtractRequest> getExtractChildRequestsForNode(ExtractRequest parentRequest)

resetExtractRequest

public void resetExtractRequest(OutgoingBatch batch)

Specified by:

resetExtractRequest in interface IDataExtractorService

requestExtractRequest

public ExtractRequest requestExtractRequest(ISqlTransaction transaction,
 String nodeId,
 String queue,
 TriggerRouter triggerRouter,
 long startBatchId,
 long endBatchId,
 long loadId,
 String table,
 long rows,
 long parentRequestId)

Specified by:

requestExtractRequest in interface IDataExtractorService

updateExtractRequestStatus

protected void updateExtractRequestStatus(ISqlTransaction transaction,
 long extractId,
 ExtractRequest.ExtractStatus status,
 long extractedRows,
 long extractedMillis)

canProcessExtractRequest

protected boolean canProcessExtractRequest(ExtractRequest request,
 NodeCommunication.CommunicationType communicationType)

execute

public void execute(NodeCommunication nodeCommunication,
 RemoteNodeStatus status)

This is a callback method used by the NodeCommunicationService that extracts an initial load in the background.

Specified by:

execute in interface INodeCommunicationService.INodeCommunicationExecutor

restartExtractRequest

protected void restartExtractRequest(List<OutgoingBatch> batches,
 ExtractRequest request,
 List<ExtractRequest> childRequests)

releaseMissedExtractRequests

public void releaseMissedExtractRequests()

Specified by:

releaseMissedExtractRequests in interface IDataExtractorService

checkSendDeferredConstraints

protected void checkSendDeferredConstraints(ExtractRequest request,
 List<ExtractRequest> childRequests,
 Node targetNode)

isApplicable

protected boolean isApplicable(NodeCommunication nodeCommunication)

buildMultiBatchStagingWriter

protected MultiBatchStagingWriter buildMultiBatchStagingWriter(ExtractRequest request,
 List<ExtractRequest> childRequests,
 Node sourceNode,
 Node targetNode,
 List<OutgoingBatch> batches,
 ProcessInfo processInfo,
 Channel channel,
 boolean isRestarted)

getProcessType

protected ProcessType getProcessType()

removeBatchFromStaging

public void removeBatchFromStaging(OutgoingBatch batch)

Specified by:

removeBatchFromStaging in interface IDataExtractorService

updateExtractRequestStatuses

public void updateExtractRequestStatuses(ISqlTransaction transaction,
 long loadId,
 String sourceNodeId,
 String fromStatus,
 String toStatus)

Specified by:

updateExtractRequestStatuses in interface IDataExtractorService