metaflow

README

metaflow

一款简单的多源数据流元数据封装

使用说明

package main

import (
	"github.com/OpenListTeam/metaflow"
	_ "github.com/OpenListTeam/metaflow/file"
)

func main() {
	streamMetadata := &metaflow.StreamMetadata{
		URL: "file://temp1/tmep2/test.txt",
	}
	content := []byte("Hello, Metaflow!")
	write(streamMetadata, content)
	result := readAll(streamMetadata)
	println(string(result))
}

func readAll(streamMetadata *metaflow.StreamMetadata) []byte {
	streamFlow, err := metaflow.CreateStream(streamMetadata)
	if err != nil {
		panic(err)
	}
	defer streamFlow.Close()

	content := make([]byte, streamMetadata.Size)
	_, err = streamFlow.Read(content)
	if err != nil {
		panic(err)
	}
	return content
}

func write(streamMetadata *metaflow.StreamMetadata, content []byte) {
	streamFlow, err := metaflow.CreateStream(streamMetadata)
	if err != nil {
		panic(err)
	}
	defer streamFlow.Close()
	_, err = streamFlow.Write(content)
	if err != nil {
		panic(err)
	}
}

贡献

欢迎提交 PR 或 Issue，帮助我们改进 Metaflow。请确保您的代码遵循 Go 语言的编码规范，并添加必要的注释和文档。

示例

package xxx

import (
	"github.com/OpenListTeam/metaflow"
)

func init(){
	metaflow.RegsitryFactoryBuilder("xxx", func(meta *metaflow.StreamMetadata) (metaflow.Stream, error) {
		return newXXXStreamProcessor(meta)
	})
}

func newXXXStreamProcessor(metadata *metaflow.StreamMetadata) (metaflow.Stream, error) {
	processor := &xxxStreamProcessor{
		metadata: metadata,
	}
	return processor, nil
}

// xxxStreamProcessor 实现了 metaflow.Stream 接口
...

Documentation

Index

• func RegisterFactoryBuilder(schema string, creator CreateFunc) error
• type CreateFunc
• type Stream
  • func CreateStream(metadata *StreamMetadata) (Stream, error)
• type StreamMetadata

Functions

func RegisterFactoryBuilder

func RegisterFactoryBuilder(schema string, creator CreateFunc) error

Types

type CreateFunc

type CreateFunc func(meta *StreamMetadata) (Stream, error)

type Stream

type Stream interface {
	io.Reader
	io.ReaderAt
	io.Writer
	io.Closer
	io.Seeker

	GetMetadata() *StreamMetadata                        // 获取元数据
	Checksum() (string, error)                           // 计算内容校验和
	PartialChecksum(offset, limit int64) (string, error) // 计算内容校验和
	IsReadOnly() bool                                    // 判断是否为只读流
}

统一的流处理器接口

func CreateStream

func CreateStream(metadata *StreamMetadata) (Stream, error)

创建流处理器

type StreamMetadata

type StreamMetadata struct {
	URL       string            `json:"url"`                  // 资源URL
	Size      int64             `json:"size,omitempty"`       // 流大小（可选）
	Checksum  string            `json:"checksum,omitempty"`   // 内容校验和（可选）
	ExpiresAt time.Time         `json:"expires_at,omitempty"` // 过期时间（可选）
	Metadata  map[string]string `json:"metadata,omitempty"`   // 扩展元数据
}

流元数据结构