ftpreader

Documentation

Index

• Constants
• type FtpHelper
• type FtpReaderJob
  • func NewFtpReaderJob() *FtpReaderJob
  • func (job *FtpReaderJob) Destroy() error
  • func (job *FtpReaderJob) Init(config config.Configuration) error
  • func (job *FtpReaderJob) Post() error
  • func (job *FtpReaderJob) Prepare() error
  • func (job *FtpReaderJob) Split(adviceNumber int) ([]config.Configuration, error)
• type FtpReaderJobFactory
  • func (f *FtpReaderJobFactory) CreateReaderJob() plugin.ReaderJob
• type FtpReaderTask
  • func NewFtpReaderTask() *FtpReaderTask
  • func (task *FtpReaderTask) Destroy() error
  • func (task *FtpReaderTask) Init(config config.Configuration) error
  • func (task *FtpReaderTask) Post() error
  • func (task *FtpReaderTask) Prepare() error
  • func (task *FtpReaderTask) StartRead(recordSender plugin.RecordSender) error
• type FtpReaderTaskFactory
  • func (f *FtpReaderTaskFactory) CreateReaderTask() plugin.ReaderTask
• type SftpHelper
  • func NewSftpHelper() *SftpHelper
  • func (h *SftpHelper) GetAllFiles(paths []string, currentLevel, maxLevel int) ([]string, error)
  • func (h *SftpHelper) GetInputStream(fileName string) (io.ReadCloser, error)
  • func (h *SftpHelper) LoginFtpServer(host, username, password string, port, timeout int, connectPattern string) error
  • func (h *SftpHelper) LogoutFtpServer() error
• type StandardFtpHelper
  • func NewStandardFtpHelper() *StandardFtpHelper
  • func (h *StandardFtpHelper) GetAllFiles(paths []string, currentLevel, maxLevel int) ([]string, error)
  • func (h *StandardFtpHelper) GetInputStream(fileName string) (io.ReadCloser, error)
  • func (h *StandardFtpHelper) LoginFtpServer(host, username, password string, port, timeout int, connectPattern string) error
  • func (h *StandardFtpHelper) LogoutFtpServer() error

Constants

const (
	KeyProtocol          = "protocol"
	KeyHost              = "host"
	KeyUsername          = "username"
	KeyPassword          = "password"
	KeyPort              = "port"
	KeyTimeout           = "timeout"
	KeyConnectPattern    = "connectPattern"
	KeyPath              = "path"
	KeyMaxTraversalLevel = "maxTraversalLevel"
	KeyFieldDelimiter    = "fieldDelimiter"
	KeyEncoding          = "encoding"
	KeyCompress          = "compress"
	KeySkipHeader        = "skipHeader"
	KeyNullFormat        = "nullFormat"
	KeyCsvReaderConfig   = "csvReaderConfig"
	KeyColumn            = "column"
)

配置键常量

const (
	DefaultFtpPort           = 21
	DefaultSftpPort          = 22
	DefaultTimeout           = 60000 // 毫秒
	DefaultMaxTraversalLevel = 100
	DefaultFtpConnectPattern = "PASV"
	DefaultFieldDelimiter    = ","
	DefaultEncoding          = "UTF-8"
	DefaultNullFormat        = "\\N"
)

默认值常量

const (
	SourceFiles = "sourceFiles"
)

内部使用常量

Types

type FtpHelper

type FtpHelper interface {
	LoginFtpServer(host, username, password string, port, timeout int, connectPattern string) error
	LogoutFtpServer() error
	GetAllFiles(paths []string, currentLevel, maxLevel int) ([]string, error)
	GetInputStream(fileName string) (io.ReadCloser, error)
}

FtpHelper 接口定义了FTP和SFTP操作的通用方法

type FtpReaderJob

type FtpReaderJob struct {
	// contains filtered or unexported fields
}

FtpReaderJob FTP读取作业，复制Java版本的功能

func NewFtpReaderJob

func NewFtpReaderJob() *FtpReaderJob

func (*FtpReaderJob) Destroy

func (job *FtpReaderJob) Destroy() error

func (*FtpReaderJob) Init

func (job *FtpReaderJob) Init(config config.Configuration) error

func (*FtpReaderJob) Post

func (job *FtpReaderJob) Post() error

func (*FtpReaderJob) Prepare

func (job *FtpReaderJob) Prepare() error

func (*FtpReaderJob) Split

func (job *FtpReaderJob) Split(adviceNumber int) ([]config.Configuration, error)

type FtpReaderJobFactory

type FtpReaderJobFactory struct{}

FtpReaderJobFactory 实现ReaderJobFactory接口

func (*FtpReaderJobFactory) CreateReaderJob

func (f *FtpReaderJobFactory) CreateReaderJob() plugin.ReaderJob

type FtpReaderTask

type FtpReaderTask struct {
	// contains filtered or unexported fields
}

FtpReaderTask FTP读取任务

func NewFtpReaderTask

func NewFtpReaderTask() *FtpReaderTask

func (*FtpReaderTask) Destroy

func (task *FtpReaderTask) Destroy() error

func (*FtpReaderTask) Init

func (task *FtpReaderTask) Init(config config.Configuration) error

func (*FtpReaderTask) Post

func (task *FtpReaderTask) Post() error

func (*FtpReaderTask) Prepare

func (task *FtpReaderTask) Prepare() error

func (*FtpReaderTask) StartRead

func (task *FtpReaderTask) StartRead(recordSender plugin.RecordSender) error

type FtpReaderTaskFactory

type FtpReaderTaskFactory struct{}

FtpReaderTaskFactory 实现ReaderTaskFactory接口

func (*FtpReaderTaskFactory) CreateReaderTask

func (f *FtpReaderTaskFactory) CreateReaderTask() plugin.ReaderTask

type SftpHelper

type SftpHelper struct {
	// contains filtered or unexported fields
}

SftpHelper 实现SFTP协议

func NewSftpHelper

func NewSftpHelper() *SftpHelper

func (*SftpHelper) GetAllFiles

func (h *SftpHelper) GetAllFiles(paths []string, currentLevel, maxLevel int) ([]string, error)

func (*SftpHelper) GetInputStream

func (h *SftpHelper) GetInputStream(fileName string) (io.ReadCloser, error)

func (*SftpHelper) LoginFtpServer

func (h *SftpHelper) LoginFtpServer(host, username, password string, port, timeout int, connectPattern string) error

func (*SftpHelper) LogoutFtpServer

func (h *SftpHelper) LogoutFtpServer() error

type StandardFtpHelper

type StandardFtpHelper struct {
	// contains filtered or unexported fields
}

StandardFtpHelper 实现标准FTP协议

func NewStandardFtpHelper

func NewStandardFtpHelper() *StandardFtpHelper

func (*StandardFtpHelper) GetAllFiles

func (h *StandardFtpHelper) GetAllFiles(paths []string, currentLevel, maxLevel int) ([]string, error)

func (*StandardFtpHelper) GetInputStream

func (h *StandardFtpHelper) GetInputStream(fileName string) (io.ReadCloser, error)

func (*StandardFtpHelper) LoginFtpServer

func (h *StandardFtpHelper) LoginFtpServer(host, username, password string, port, timeout int, connectPattern string) error

func (*StandardFtpHelper) LogoutFtpServer

func (h *StandardFtpHelper) LogoutFtpServer() error