开发具有同步输出的自定义转换组件

组件的输入、输出和自定义属性通常在 ProvideComponentProperties 方法期间创建。您可以使用两种方式添加具有同步输出的转换组件的输入和输出。可以使用方法的基类实现，然后修改该方法创建的默认输入和输出，也可以自己显式添加输入和输出。

下面的代码示例演示显式添加输入和输出对象的 ProvideComponentProperties 实现。注释中包含对将要完成同样任务的基类的调用。

using Microsoft.SqlServer.Dts.Pipeline;
using Microsoft.SqlServer.Dts.Pipeline.Wrapper;
using Microsoft.SqlServer.Dts.Runtime;

namespace Microsoft.Samples.SqlServer.Dts
{
    [DtsPipelineComponent(DisplayName = "SynchronousComponent", ComponentType = ComponentType.Transform)]
    public class SyncComponent : PipelineComponent
    {

        public override void ProvideComponentProperties()
        {
            // Add the input.
            IDTSInput100 input = ComponentMetaData.InputCollection.New();
            input.Name = "Input";

            // Add the output.
            IDTSOutput100 output = ComponentMetaData.OutputCollection.New();
            output.Name = "Output";
            output.SynchronousInputID = input.ID;

            // Alternatively, you can let the base class add the input and output
            // and set the SynchronousInputID of the output to the ID of the input.
            // base.ProvideComponentProperties();
        }
    }
}

Imports Microsoft.SqlServer.Dts.Pipeline
Imports Microsoft.SqlServer.Dts.Pipeline.Wrapper
Imports Microsoft.SqlServer.Dts.Runtime

<DtsPipelineComponent(DisplayName:="SynchronousComponent", ComponentType:=ComponentType.Transform)> _
Public Class SyncComponent
    Inherits PipelineComponent

    Public Overrides Sub ProvideComponentProperties()

        ' Add the input.
        Dim input As IDTSInput100 = ComponentMetaData.InputCollection.New()
        input.Name = "Input"

        ' Add the output.
        Dim output As IDTSOutput100 = ComponentMetaData.OutputCollection.New()
        output.Name = "Output"
        output.SynchronousInputID = Input.ID

        ' Alternatively, you can let the base class add the input and output
        ' and set the SynchronousInputID of the output to the ID of the input.
        ' base.ProvideComponentProperties();

    End Sub

End Class

尽管具有同步输出的转换组件不向缓冲区添加行，但是它们可能向自己的输出添加额外的输出列。通常，当组件添加输出列时，新输出列的值在运行时从一个或多个列中包含的数据派生而来，这些列由上游组件提供给该组件。

创建输出列后，必须设置该列的数据类型属性。输出列的数据类型属性的设置需要特殊处理，并通过调用 SetDataTypeProperties 方法来执行。此方法是必需的，因为 DataType、Length、Precision 和 CodePage 属性是只读的，并且每个属性都依赖于其他属性的设置。此方法可保证属性值的设置保持一致，并且数据流任务会验证是否正确设置了这些值。

列的 DataType 决定了为其他属性设置的值。下表说明了对每个 DataType 的依赖属性的要求。未列出的数据类型的依赖属性设置为零。

由于对数据类型属性的限制是基于输出列的数据类型的，因此使用托管类型时必须选择正确的 Integration Services 数据类型。基类提供了三个帮助器方法：ConvertBufferDataTypeToFitManaged、BufferTypeToDataRecordType 和 DataRecordTypeToBufferType，它们可协助托管组件开发人员在给定托管类型的情况下选择 SSIS 数据类型。这些方法用于将托管数据类型转换为 SSIS 数据类型，反之亦然。

在执行期间提供给组件的缓冲区中的列数将很可能超过组件的输入或输出集合中的列数。这是因为每个缓冲区都包含在数据流的组件中定义的所有输出列。为了确保缓冲区列与输出或输出的列正确匹配，组件开发人员必须使用 BufferManager 的 FindColumnByLineageID 方法。此方法会根据其沿袭 ID 在已定义的缓冲区中查找列。这些列通常在 PreExecute 期间进行查找，因为这是第一个其 BufferManager 属性可用的运行时方法。

下面的代码示例演示了一个组件，该组件在 PreExecute 期间查找其输入和输出列集合中的列的索引。列的索引存储在整数数组中，可在 ProcessInput 期间由组件访问。

int []inputColumns;
int []outputColumns;

public override void PreExecute()
{
    IDTSInput100 input = ComponentMetaData.InputCollection[0];
    IDTSOutput100 output = ComponentMetaData.OutputCollection[0];

    inputColumns = new int[input.InputColumnCollection.Count];
    outputColumns = new int[output.OutputColumnCollection.Count];

    for(int col=0; col < input.InputColumnCollection.Count; col++)
    {
        IDTSInputColumn100 inputColumn = input.InputColumnCollection[col];
        inputColumns[col] = BufferManager.FindColumnByLineageID(input.Buffer, inputColumn.LineageID);
    }

    for(int col=0; col < output.OutputColumnCollection.Count; col++)
    {
        IDTSOutputColumn100 outputColumn = output.OutputColumnCollection[col];
        outputColumns[col] = BufferManager.FindColumnByLineageID(input.Buffer, outputColumn.LineageID);
    }

}

Public Overrides Sub PreExecute()

    Dim input As IDTSInput100 = ComponentMetaData.InputCollection(0)
    Dim output As IDTSOutput100 = ComponentMetaData.OutputCollection(0)


    ReDim inputColumns(input.InputColumnCollection.Count)
    ReDim outputColumns(output.OutputColumnCollection.Count)

    For col As Integer = 0 To input.InputColumnCollection.Count

        Dim inputColumn As IDTSInputColumn100 = input.InputColumnCollection(col)
        inputColumns(col) = BufferManager.FindColumnByLineageID(input.Buffer, inputColumn.LineageID)
    Next

    For col As Integer = 0 To output.OutputColumnCollection.Count

        Dim outputColumn As IDTSOutputColumn100 = output.OutputColumnCollection(col)
        outputColumns(col) = BufferManager.FindColumnByLineageID(input.Buffer, outputColumn.LineageID)
    Next

End Sub

组件可接收 PipelineBuffer 对象，这些对象包含 ProcessInput 方法中的行和列。在此方法中，将循环访问缓冲区中的行，并读取和修改 PreExecute 期间标识的列。数据流任务将重复调用该方法，直到上游组件不再提供行。

可使用数组索引器访问方法或者使用 Get 或 Set 方法之一来读取或写入缓冲区中的单个列。Get 和 Set 方法的效率更高，应在缓冲区中的列的数据类型已知时使用。

下面的代码示例演示 ProcessInput 方法的实现，该方法用于处理传入的行。

public override void ProcessInput( int InputID, PipelineBuffer buffer)
{
       while( buffer.NextRow())
       {
            for(int x=0; x < inputColumns.Length;x++)
            {
                if(!buffer.IsNull(inputColumns[x]))
                {
                    object columnData = buffer[inputColumns[x]];
                    // TODO: Modify the column data.
                    buffer[inputColumns[x]] = columnData;
                }
            }
        
      }
}

Public Overrides Sub ProcessInput(ByVal InputID As Integer, ByVal buffer As PipelineBuffer)

        While (buffer.NextRow())

            For x As Integer = 0 To inputColumns.Length

                if buffer.IsNull(inputColumns(x)) = false then

                    Dim columnData As Object = buffer(inputColumns(x))
                    ' TODO: Modify the column data.
                    buffer(inputColumns(x)) = columnData

                End If
            Next

        End While
End Sub