]]> _
_
Public Class ScriptMain
Inherits UserComponent
Private JobUrl_3P As String = "{0}://{1}/services/async/{2}/job/"
Private BulkNS As String = "http://www.force.com/2009/06/asyncapi/dataload"
'''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''
Public Overrides Sub Input_ProcessInputRow(ByVal Row As InputBuffer)
Dim batchId As String = CStr(Row.Buffer(m_batchIdx))
Try
If m_batchId <> batchId Then
' Load next batch.
m_batchId = batchId
If String.IsNullOrEmpty(m_jobId) Then
' Load job identifier.
m_jobId = CStr(GetVariable_(Me.JobIdVariable))
End If
If Not m_batchRes Is Nothing Then
Call m_batchRes.Dispose()
End If
m_batchRes = GetBatchResult_(m_service, m_jobId, m_batchId)
End If
' Get batch result.
Dim record() As String = m_batchRes.ReadFields()
If record Is Nothing Then
' There is no more data.
Exit Sub
End If
With Row
.ResultId = record(0)
.Success = Convert.ToBoolean(record(1))
.Created = Convert.ToBoolean(record(2))
.Error = record(3)
End With
Catch ex As Exception
Call FireError_(ex.ToString())
End Try
End Sub ' Input_ProcessInputRow
'''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''
Public Overrides Sub PreExecute()
Call MyBase.PreExecute()
m_connection = CType( _
MyBase.Connections.Connection.AcquireConnection(Nothing), _
ISforceConnection)
m_service = CType(m_connection.Connect(), SforceService)
Dim input As IDTSInput90 = Me.ComponentMetaData.InputCollection(0)
' Get batch column index.
Dim inputCol As IDTSInputColumn90 = input.InputColumnCollection(Me.BatchIdColumn)
m_batchIdx = Me.HostComponent.BufferManager.FindColumnByLineageID(input.Buffer, inputCol.LineageID)
End Sub ' PreExecute
'''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''
Public Overrides Sub PostExecute()
Call MyBase.PostExecute()
Call m_connection.Close()
End Sub ' PostExecute
'''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''
Public Function Validate(ByRef errMessage As String) As Boolean
Dim result As Boolean
Try
If String.IsNullOrEmpty(Me.Connection) Then
Throw New ApplicationException("Select Salesforce connection.")
End If
If String.IsNullOrEmpty(Me.JobIdVariable) Then
Throw New ApplicationException("Select job id variable.")
End If
If String.IsNullOrEmpty(Me.BatchIdColumn) Then
Throw New ApplicationException("Select batch id column.")
End If
' Setup batch id column.
Dim input As IDTSInput90 = MyBase.ComponentMetaData.InputCollection(0)
Call input.InputColumnCollection.RemoveAll()
Dim virtInput As IDTSVirtualInput90 = input.GetVirtualInput()
Dim virtColumn As IDTSVirtualInputColumn90 = virtInput.VirtualInputColumnCollection(Me.BatchIdColumn)
Call virtInput.SetUsageType(virtColumn.LineageID, DTSUsageType.UT_READONLY)
' Store connection information in the runtime connection collection, too.
' Cannot directly use RuntimeConnectionCollection in the property get/set
' because of issues with multi-threading.
Me.ComponentMetaData.RuntimeConnectionCollection("Connection").ConnectionManagerID = Me.Connection
result = True
Catch ex As Exception
result = False
errMessage = ex.Message
End Try
Validate = result
End Function 'Validate
#Region "Properties"
'''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''
_
_
_
Public Property Connection() As String
Get
Connection = m_sforceConnection
End Get
Set(ByVal value As String)
m_sforceConnection = value
End Set
End Property ' Connection
'''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''
_
_
_
Public Property JobIdVariable() As String
Get
JobIdVariable = m_jobIdVar
End Get
Set(ByVal value As String)
m_jobIdVar = value
End Set
End Property ' JobIdVariable
'''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''
_
_
_
Public Property BatchIdColumn() As String
Get
BatchIdColumn = m_batchIdCol
End Get
Set(ByVal value As String)
m_batchIdCol = value
End Set
End Property ' BatchIdColumn
#End Region ' Properties
#Region "Internals"
'''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''
Private ReadOnly Property ConnectionType() As String()
Get
ConnectionType = New String() {"SFORCE"}
End Get
End Property ' ConnectionType
'''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''
Private Sub FireError_(ByVal message As String)
Dim cancel As Boolean = False
Call MyBase.ComponentMetaData.FireError( _
0, _
"Salesforce Bulk Result", _
message, _
String.Empty, _
0, _
cancel)
End Sub ' FireError_
'''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''
Private Function GetVariable_(ByVal varName As String) As Object
Dim result As Object
Dim vars As IDTSVariables90 = Nothing
Call Me.VariableDispenser.LockOneForRead(varName, vars)
Try
result = vars(varName).Value
Finally
Call vars.Unlock()
End Try
GetVariable_ = result
End Function ' GetVariable_
'''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''
' Find the API version in use by stepping the service URL from the back.
' The first segment which can be converted to number is the version.
Private Function GetServiceVersion_(ByVal url As Uri) As Double
Dim result As Double = 27
Dim segments As String() = url.Segments
Dim segsCount As Integer = segments.Length
For segIndex As Integer = segsCount - 1 To 0 Step -1
Dim segment As String = segments(segIndex).Trim("/"c)
If Double.TryParse(segment, NumberStyles.Any, CultureInfo.InvariantCulture, result) Then
Exit For
End If
Next
GetServiceVersion_ = result
End Function ' GetServiceVersion_
'''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''
Private Function PrepareRequest_( _
ByVal service As SforceService, _
ByVal url As String) As HttpWebRequest
' Setup base job request url.
Dim serviceUri As Uri = New Uri(service.Url)
Dim jobUrl As String = String.Format( _
JobUrl_3P, _
serviceUri.Scheme, _
serviceUri.Host, _
GetServiceVersion_(serviceUri))
Dim result As HttpWebRequest = CType(WebRequest.Create(jobUrl + url), HttpWebRequest)
Call result.Headers.Add("X-SFDC-Session", service.SessionHeaderValue.sessionId)
PrepareRequest_ = result
End Function ' PrepareRequest_
'''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''
Private Function GetXmlResult_(ByVal req As HttpWebRequest) As XmlDocument
Dim result As New XmlDocument
Try
Using res As WebResponse = req.GetResponse()
Call result.Load(res.GetResponseStream())
End Using
Catch ex As WebException
' The regular exception information is not very useful. Extract error information.
If ex.Response Is Nothing Then
' Error happened before server was able to respond.
Throw
End If
Using errorStream As Stream = ex.Response.GetResponseStream()
Throw New Exception(New StreamReader(errorStream).ReadToEnd(), ex)
End Using
End Try
GetXmlResult_ = result
End Function ' GetXmlResult_
'''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''
Private Function GetCsvResult_(ByVal req As HttpWebRequest) As TextFieldParser
Dim result As TextFieldParser
Try
m_readRes = req.GetResponse()
result = New TextFieldParser(m_readRes.GetResponseStream())
result.HasFieldsEnclosedInQuotes = True
Call result.SetDelimiters(",")
' Skip first row which is header.
Call result.ReadFields()
Catch ex As WebException
' The regular exception information is not very useful. Extract error information.
If ex.Response Is Nothing Then
' Error happened before server was able to respond.
Throw
End If
Using errorStream As Stream = ex.Response.GetResponseStream()
Throw New Exception(New StreamReader(errorStream).ReadToEnd(), ex)
End Using
End Try
GetCsvResult_ = result
End Function ' GetCsvResult_
'''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''
Private Function GetBatchResult_( _
ByVal service As SforceService, _
ByVal jobId As String, _
ByVal batchId As String) As TextFieldParser
' Loop until batch result is ready.
While True
Dim doc As XmlDocument = GetXmlResult_(PrepareRequest_(service, jobId + "/batch/" + batchId))
Dim ns As XmlNamespaceManager = New XmlNamespaceManager(doc.NameTable)
Call ns.AddNamespace("bl", BulkNS)
Dim state As String = doc.SelectSingleNode("/bl:batchInfo/bl:state", ns).InnerText
Select Case state
Case "Completed"
' Ready.
Exit While
Case "InProgress", "Queued"
' Have to wait.
Case "Failed", "Not Processed"
Throw New Exception(String.Format("'{0}' batch failed or aborted.", batchId))
End Select
' Wait 5 seconds and try again.
Call Thread.Sleep(5 * 1000)
End While
GetBatchResult_ = GetCsvResult_(PrepareRequest_(service, jobId + "/batch/" + batchId + "/result"))
End Function ' GetBatchResult_
#End Region ' Internals
#Region "Attributes"
Private m_sforceConnection As String
Private m_jobIdVar As String
Private m_batchIdCol As String
Private m_connection As ISforceConnection
Private m_service As SforceService
Private m_batchIdx As Integer
Private m_jobId As String
Private m_batchId As String
Private m_readRes As WebResponse
Private m_batchRes As TextFieldParser
#End Region 'Attributes
End Class ' ScriptMain
]]>ScriptComponent_e6007d71caeb4ae18f1e3ca85dc10c00trueCozyRoc.SqlServer.SSIS.ScriptComponentHostPlus, CozyRoc.SSISPlus, Version=1.0.0.0, Culture=neutral, PublicKeyToken=16cf490bb80c34ea