kafkafs

Documentation

Index

• type KafkaClient
  • func NewKafkaClient(client *sarama.Client, maxBytes int32) KafkaClient
• type KafkaRoFs
  • func NewKafkaRoFs(kClient KafkaClient) *KafkaRoFs
  • func (fs *KafkaRoFs) GetAttr(name string, context *fuse.Context) (*fuse.Attr, fuse.Status)
  • func (fs *KafkaRoFs) Open(name string, flags uint32, context *fuse.Context) (nodefs.File, fuse.Status)
  • func (fs *KafkaRoFs) OpenDir(name string, context *fuse.Context) ([]fuse.DirEntry, fuse.Status)
  • func (fs *KafkaRoFs) Unlink(name string, context *fuse.Context) fuse.Status
  • func (fs *KafkaRoFs) Utimens(name string, Atime *time.Time, Mtime *time.Time, context *fuse.Context) fuse.Status

Types

type KafkaClient

type KafkaClient interface {
	// list all topics
	GetTopics() ([]string, error)

	// list the partitions for a topic
	GetPartitions(topic string) ([]int32, error)

	// get the earliest and next offsets available for a topic /
	// partition.  Note that the "next" offset is for the next
	// message, that is, the one that currently doesn't exist yet
	// (e.g., 0 if the partition is empty)
	GetBoundingOffsets(topic string, partition int32) (int64, int64, error)

	// get the bytes for a given message
	GetMessage(topic string, partition int32, offset int64) ([]byte, error)
}

A much simplified interface for dealing with Kafka, restricted only to the operations that kafkafs needs and therefore easier to mock / control.

func NewKafkaClient

func NewKafkaClient(client *sarama.Client, maxBytes int32) KafkaClient

type KafkaRoFs

type KafkaRoFs struct {
	pathfs.FileSystem

	KafkaClient KafkaClient
	// contains filtered or unexported fields
}

Implements a FUSE filesystem backed by a Kafka installation.

This version is read only, so it cannot post to topics, only read from them.

func NewKafkaRoFs

func NewKafkaRoFs(kClient KafkaClient) *KafkaRoFs

func (*KafkaRoFs) GetAttr

func (fs *KafkaRoFs) GetAttr(name string, context *fuse.Context) (*fuse.Attr,
	fuse.Status)

func (*KafkaRoFs) Open

func (fs *KafkaRoFs) Open(name string, flags uint32,
	context *fuse.Context) (nodefs.File, fuse.Status)

func (*KafkaRoFs) OpenDir

func (fs *KafkaRoFs) OpenDir(name string, context *fuse.Context) ([]fuse.DirEntry,
	fuse.Status)

func (*KafkaRoFs) Unlink

func (fs *KafkaRoFs) Unlink(name string, context *fuse.Context) fuse.Status

func (*KafkaRoFs) Utimens

func (fs *KafkaRoFs) Utimens(name string, Atime *time.Time, Mtime *time.Time,
	context *fuse.Context) fuse.Status

just pretend we set the times to keep touch happy