init

.gitignore

@@ -10,5 +10,9 @@
 # Output of the go coverage tool, specifically when used with LiteIDE
 *.out
 
-# Project-local glide cache, RE: https://github.com/Masterminds/glide/issues/736
-.glide/
+# Self
+bin/
+packages/
+
+# Cache
+*.swp

Makefile

@@ -0,0 +1,15 @@
+all: fmt build
+
+build: fft fftw ffts
+
+fmt:
+	go fmt ./...
+
+fft:
+	go build -ldflags "-s -w" -o bin/fft ./cmd/fft
+
+fftw:
+	go build -ldflags "-s -w" -o bin/fftw ./cmd/fftw
+
+ffts:
+	go build -ldflags "-s -w" -o bin/ffts ./cmd/ffts

Makefile.cross-compiles

@@ -0,0 +1,41 @@
+export PATH := $(GOPATH)/bin:$(PATH)
+LDFLAGS := -s -w
+
+all: build
+
+build: app
+
+app:
+	env CGO_ENABLED=0 GOOS=darwin GOARCH=amd64 go build -ldflags "$(LDFLAGS)" -o ./fft_darwin_amd64 ./cmd/fft
+	env CGO_ENABLED=0 GOOS=darwin GOARCH=amd64 go build -ldflags "$(LDFLAGS)" -o ./fftw_darwin_amd64 ./cmd/fftw
+	env CGO_ENABLED=0 GOOS=darwin GOARCH=amd64 go build -ldflags "$(LDFLAGS)" -o ./ffts_darwin_amd64 ./cmd/ffts
+	env CGO_ENABLED=0 GOOS=freebsd GOARCH=amd64 go build -ldflags "$(LDFLAGS)" -o ./fft_freebsd_amd64 ./cmd/fft
+	env CGO_ENABLED=0 GOOS=freebsd GOARCH=amd64 go build -ldflags "$(LDFLAGS)" -o ./fftw_freebsd_amd64 ./cmd/fftw
+	env CGO_ENABLED=0 GOOS=freebsd GOARCH=amd64 go build -ldflags "$(LDFLAGS)" -o ./ffts_freebsd_amd64 ./cmd/ffts
+	env CGO_ENABLED=0 GOOS=linux GOARCH=386 go build -ldflags "$(LDFLAGS)" -o ./fft_linux_386 ./cmd/fft
+	env CGO_ENABLED=0 GOOS=linux GOARCH=386 go build -ldflags "$(LDFLAGS)" -o ./fftw_linux_386 ./cmd/fftw
+	env CGO_ENABLED=0 GOOS=linux GOARCH=386 go build -ldflags "$(LDFLAGS)" -o ./ffts_linux_386 ./cmd/ffts
+	env CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -ldflags "$(LDFLAGS)" -o ./fft_linux_amd64 ./cmd/fft
+	env CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -ldflags "$(LDFLAGS)" -o ./fftw_linux_amd64 ./cmd/fftw
+	env CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -ldflags "$(LDFLAGS)" -o ./ffts_linux_amd64 ./cmd/ffts
+	env CGO_ENABLED=0 GOOS=linux GOARCH=arm go build -ldflags "$(LDFLAGS)" -o ./fft_linux_arm ./cmd/fft
+	env CGO_ENABLED=0 GOOS=linux GOARCH=arm go build -ldflags "$(LDFLAGS)" -o ./fftw_linux_arm ./cmd/fftw
+	env CGO_ENABLED=0 GOOS=linux GOARCH=arm go build -ldflags "$(LDFLAGS)" -o ./ffts_linux_arm ./cmd/ffts
+	env CGO_ENABLED=0 GOOS=linux GOARCH=arm64 go build -ldflags "$(LDFLAGS)" -o ./fft_linux_arm64 ./cmd/fft
+	env CGO_ENABLED=0 GOOS=linux GOARCH=arm64 go build -ldflags "$(LDFLAGS)" -o ./fftw_linux_arm64 ./cmd/fftw
+	env CGO_ENABLED=0 GOOS=linux GOARCH=arm64 go build -ldflags "$(LDFLAGS)" -o ./ffts_linux_arm64 ./cmd/ffts
+	env CGO_ENABLED=0 GOOS=windows GOARCH=amd64 go build -ldflags "$(LDFLAGS)" -o ./fft_windows_amd64.exe ./cmd/fft
+	env CGO_ENABLED=0 GOOS=windows GOARCH=amd64 go build -ldflags "$(LDFLAGS)" -o ./fftw_windows_amd64.exe ./cmd/fftw
+	env CGO_ENABLED=0 GOOS=windows GOARCH=amd64 go build -ldflags "$(LDFLAGS)" -o ./ffts_windows_amd64.exe ./cmd/ffts
+	env CGO_ENABLED=0 GOOS=linux GOARCH=mips64 go build -ldflags "$(LDFLAGS)" -o ./fft_linux_mips64 ./cmd/fft
+	env CGO_ENABLED=0 GOOS=linux GOARCH=mips64 go build -ldflags "$(LDFLAGS)" -o ./fftw_linux_mips64 ./cmd/fftw
+	env CGO_ENABLED=0 GOOS=linux GOARCH=mips64 go build -ldflags "$(LDFLAGS)" -o ./ffts_linux_mips64 ./cmd/ffts
+	env CGO_ENABLED=0 GOOS=linux GOARCH=mips64le go build -ldflags "$(LDFLAGS)" -o ./fft_linux_mips64le ./cmd/fft
+	env CGO_ENABLED=0 GOOS=linux GOARCH=mips64le go build -ldflags "$(LDFLAGS)" -o ./fftw_linux_mips64le ./cmd/fftw
+	env CGO_ENABLED=0 GOOS=linux GOARCH=mips64le go build -ldflags "$(LDFLAGS)" -o ./ffts_linux_mips64le ./cmd/ffts
+	env CGO_ENABLED=0 GOOS=linux GOARCH=mips GOMIPS=softfloat go build -ldflags "$(LDFLAGS)" -o ./fft_linux_mips ./cmd/fft
+	env CGO_ENABLED=0 GOOS=linux GOARCH=mips GOMIPS=softfloat go build -ldflags "$(LDFLAGS)" -o ./fftw_linux_mips ./cmd/fftw
+	env CGO_ENABLED=0 GOOS=linux GOARCH=mips GOMIPS=softfloat go build -ldflags "$(LDFLAGS)" -o ./ffts_linux_mips ./cmd/ffts
+	env CGO_ENABLED=0 GOOS=linux GOARCH=mipsle GOMIPS=softfloat go build -ldflags "$(LDFLAGS)" -o ./fft_linux_mipsle ./cmd/fft
+	env CGO_ENABLED=0 GOOS=linux GOARCH=mipsle GOMIPS=softfloat go build -ldflags "$(LDFLAGS)" -o ./fftw_linux_mipsle ./cmd/fftw
+	env CGO_ENABLED=0 GOOS=linux GOARCH=mipsle GOMIPS=softfloat go build -ldflags "$(LDFLAGS)" -o ./ffts_linux_mipsle ./cmd/ffts

README.md

@@ -1 +1,37 @@
 # fft
+
+fft 是一个分布式的文件传输工具，可以同时利用多个中转节点来并行传输文件。
+
+## 开发状态
+
+目前处于早期开发阶段，功能不完善，仅用于测试使用。
+
+master 分支用于发布稳定版本，dev 分支用于开发，您可以尝试下载最新的 release 版本进行测试。
+
+**目前的交互协议可能随时改变，不保证向后兼容，升级新版本时需要注意公告说明。**
+
+## 使用示例
+
+* ffts: server 控制节点，部署一个。
+* fftw: worker 节点，负责中转流量，部署任意多个，更多的 worker 节点可以提高传输文件的速度。
+* fft: 客户端，用于发送和接收文件。
+
+每一个程序都可以通过 `-h` 来查看使用参数的说明。
+
+ffts 和 fftw 需要部署在有公网 IP 的机器上，且开放对应的端口供 fft 访问。
+
+### 发送文件
+
+`./fft -i 123 -l ./filename`
+
+`-i 123` 指定这次传输请求的 ID，需要是一个和其他人不重复的自定义值，之后将在这个 ID 通知接收方，接收方通过此 ID 来接收文件。
+
+`-l ./filename` 指定需要传输的本地文件路径。
+
+### 接收文件
+
+`./fft -i 123 -t ./`
+
+`-i 123` 指定这次接收传输请求的 ID。
+
+`-t ./` 指定保存文件到本地的路径，如果是目录，则保存发送方的文件名到指定目录，否则会创建一个新的文件。

client/recv.go

@@ -0,0 +1,123 @@
+package client
+
+import (
+	"fmt"
+	"net"
+	"os"
+	"path/filepath"
+	"time"
+
+	"github.com/fatedier/fft/pkg/msg"
+	"github.com/fatedier/fft/pkg/receiver"
+	"github.com/fatedier/fft/pkg/stream"
+)
+
+func (svc *Service) recvFile(id string, filePath string) error {
+	isDir := false
+	finfo, err := os.Stat(filePath)
+	if err == nil && finfo.IsDir() {
+		isDir = true
+	}
+
+	conn, err := net.Dial("tcp", svc.serverAddr)
+	if err != nil {
+		return err
+	}
+	defer conn.Close()
+
+	msg.WriteMsg(conn, &msg.ReceiveFile{
+		ID: id,
+	})
+
+	conn.SetReadDeadline(time.Now().Add(10 * time.Second))
+	raw, err := msg.ReadMsg(conn)
+	if err != nil {
+		return err
+	}
+	conn.SetReadDeadline(time.Time{})
+
+	m, ok := raw.(*msg.ReceiveFileResp)
+	if !ok {
+		return fmt.Errorf("get send file response format error")
+	}
+	if m.Error != "" {
+		return fmt.Errorf(m.Error)
+	}
+
+	if len(m.Workers) == 0 {
+		return fmt.Errorf("no available workers")
+	}
+	fmt.Printf("Recv filename: %s\n", m.Name)
+	if svc.debugMode {
+		fmt.Printf("Workers: %v\n", m.Workers)
+	}
+
+	realPath := filePath
+	if isDir {
+		realPath = filepath.Join(filePath, m.Name)
+	}
+	f, err := os.Create(realPath)
+	if err != nil {
+		return err
+	}
+
+	recv := receiver.NewReceiver(0, f)
+	for _, worker := range m.Workers {
+		addr := worker
+		go newRecvStream(recv, id, addr, svc.debugMode)
+	}
+	recv.Run()
+	return nil
+}
+
+func newRecvStream(recv *receiver.Receiver, id string, addr string, debugMode bool) {
+	first := true
+	for {
+		if !first {
+			time.Sleep(3 * time.Second)
+		} else {
+			first = false
+		}
+
+		conn, err := net.Dial("tcp", addr)
+		if err != nil {
+			log(debugMode, "[%s] %v", addr, err)
+			return
+		}
+
+		msg.WriteMsg(conn, &msg.NewReceiveFileStream{
+			ID: id,
+		})
+
+		conn.SetReadDeadline(time.Now().Add(10 * time.Second))
+		raw, err := msg.ReadMsg(conn)
+		if err != nil {
+			conn.Close()
+			log(debugMode, "[%s] %v", addr, err)
+			continue
+		}
+		conn.SetReadDeadline(time.Time{})
+		m, ok := raw.(*msg.NewReceiveFileStreamResp)
+		if !ok {
+			conn.Close()
+			log(debugMode, "[%s] read NewReceiveFileStreamResp format error", addr)
+			continue
+		}
+
+		if m.Error != "" {
+			conn.Close()
+			log(debugMode, "[%s] new recv file stream error: %s", addr, m.Error)
+			continue
+		}
+		fmt.Printf("connect to worker [%s] success\n", addr)
+
+		s := stream.NewFrameStream(conn)
+		for {
+			frame, err := s.ReadFrame()
+			if err != nil {
+				return
+			}
+			recv.RecvFrame(frame)
+		}
+	}
+}

client/send.go

@@ -0,0 +1,131 @@
+package client
+
+import (
+	"fmt"
+	"net"
+	"os"
+	"sync"
+	"time"
+
+	"github.com/fatedier/fft/pkg/msg"
+	"github.com/fatedier/fft/pkg/sender"
+	"github.com/fatedier/fft/pkg/stream"
+)
+
+func (svc *Service) sendFile(id string, filePath string) error {
+	conn, err := net.Dial("tcp", svc.serverAddr)
+	if err != nil {
+		return err
+	}
+	defer conn.Close()
+
+	f, err := os.Open(filePath)
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+
+	finfo, err := f.Stat()
+	if err != nil {
+		return err
+	}
+	if finfo.IsDir() {
+		return fmt.Errorf("send file can't be a directory")
+	}
+
+	msg.WriteMsg(conn, &msg.SendFile{
+		ID:   id,
+		Name: finfo.Name(),
+	})
+
+	conn.SetReadDeadline(time.Now().Add(120 * time.Second))
+	raw, err := msg.ReadMsg(conn)
+	if err != nil {
+		return err
+	}
+	conn.SetReadDeadline(time.Time{})
+
+	m, ok := raw.(*msg.SendFileResp)
+	if !ok {
+		return fmt.Errorf("get send file response format error")
+	}
+	if m.Error != "" {
+		return fmt.Errorf(m.Error)
+	}
+
+	if len(m.Workers) == 0 {
+		return fmt.Errorf("no available workers")
+	}
+	fmt.Printf("ID: %s\n", m.ID)
+	if svc.debugMode {
+		fmt.Printf("Workers: %v\n", m.Workers)
+	}
+
+	var wait sync.WaitGroup
+	doneCh := make(chan struct{})
+	s := sender.NewSender(0, f)
+
+	for _, worker := range m.Workers {
+		wait.Add(1)
+		go func(addr string) {
+			newSendStream(doneCh, s, m.ID, addr, svc.debugMode)
+			wait.Done()
+		}(worker)
+	}
+	s.Run()
+	close(doneCh)
+	wait.Wait()
+	return nil
+}
+
+func newSendStream(doneCh chan struct{}, s *sender.Sender, id string, addr string, debugMode bool) {
+	first := true
+	for {
+		select {
+		case <-doneCh:
+			return
+		default:
+		}
+
+		if !first {
+			time.Sleep(3 * time.Second)
+		} else {
+			first = false
+		}
+
+		conn, err := net.Dial("tcp", addr)
+		if err != nil {
+			log(debugMode, "[%s] %v", addr, err)
+			continue
+		}
+
+		msg.WriteMsg(conn, &msg.NewSendFileStream{
+			ID: id,
+		})
+
+		conn.SetReadDeadline(time.Now().Add(10 * time.Second))
+		raw, err := msg.ReadMsg(conn)
+		if err != nil {
+			conn.Close()
+			log(debugMode, "[%s] %v", addr, err)
+			continue
+		}
+		conn.SetReadDeadline(time.Time{})
+		m, ok := raw.(*msg.NewSendFileStreamResp)
+		if !ok {
+			conn.Close()
+			log(debugMode, "[%s] read NewSendFileStreamResp format error", addr)
+			continue
+		}
+
+		if m.Error != "" {
+			conn.Close()
+			log(debugMode, "[%s] new send file stream error: %s", addr, m.Error)
+			continue
+		}
+		fmt.Printf("connect to worker [%s] success\n", addr)
+
+		s.HandleStream(stream.NewFrameStream(conn))
+		break
+	}
+}