started working on fileserver

file-server.go

@@ -0,0 +1,69 @@
+package main
+
+import (
+	"fmt"
+
+	"github.com/gnana997/decentralised-storage-go/p2p"
+)
+
+type FileServerOpts struct {
+	RootFolder string
+
+	Transport         p2p.Transport
+	PathTransformFunc PathTransformFunc
+}
+
+type FileServer struct {
+	FileServerOpts
+
+	store  *Store
+	quitch chan struct{}
+}
+
+func NewFileServer(opts FileServerOpts) *FileServer {
+	return &FileServer{
+		FileServerOpts: opts,
+		store: NewStore(StoreOpts{
+			Root:              opts.RootFolder,
+			PathTransformFunc: opts.PathTransformFunc,
+		}),
+		quitch: make(chan struct{}),
+	}
+}
+
+func (fs *FileServer) Stop() {
+	close(fs.quitch)
+}
+
+func (fs *FileServer) loop() {
+	defer func() {
+		if err := fs.Transport.Close(); err != nil {
+			fmt.Println(err)
+		}
+		if err := fs.Close(); err != nil {
+			fmt.Println(err)
+		}
+	}()
+	for {
+		select {
+		case msg := <-fs.Transport.Consume():
+			fmt.Println(msg)
+		case <-fs.quitch:
+			return
+		}
+	}
+}
+
+func (fs *FileServer) Start() error {
+	if err := fs.Transport.ListenAndAccept(); err != nil {
+		return err
+	}
+
+	fs.loop()
+
+	return nil
+}
+
+func (fs *FileServer) Close() error {
+	return fs.store.Close()
+}

main.go

@@ -2,7 +2,7 @@ package main
 
 import (
 	"fmt"
-	"log"
+	"time"
 
 	"github.com/gnana997/decentralised-storage-go/p2p"
 )
@@ -15,27 +15,29 @@ func OnPeer(peer p2p.Peer) error {
 
 func main() {
 
-	tcpOpts := p2p.TCPTransportOptions{
+	tcpTransportOpts := p2p.TCPTransportOptions{
 		ListenAddr:    ":3000",
 		HandshakeFunc: p2p.NOPHandshakeFunc,
 		Decoder:       p2p.NOPDecoder{},
-		OnPeer:        OnPeer,
 	}
+	tcptransport := p2p.NewTCPTransport(tcpTransportOpts)
 
-	tr := p2p.NewTCPTransport(tcpOpts)
+	fileServerOpts := FileServerOpts{
+		RootFolder:        "3000_network",
+		PathTransformFunc: CASPathTransformFunc,
+
+		Transport: tcptransport,
+	}
+
+	fs := NewFileServer(fileServerOpts)
 
 	go func() {
-		for {
-			rpc := <-tr.Consume()
-			log.Printf("received rpc: %+v", rpc)
-		}
+		time.Sleep(time.Second * 3)
+		fs.Stop()
 	}()
 
-	if err := tr.ListenAndAccept(); err != nil {
-		log.Fatal(err)
+	if err := fs.Start(); err != nil {
+		fmt.Println(err)
+		return
 	}
-
-	fmt.Println("building decentralised storage system")
-
-	select {}
 }

p2p/tcp_transport.go

@@ -1,7 +1,9 @@
 package p2p
 
 import (
+	"errors"
 	"fmt"
+	"log"
 	"net"
 	"sync"
 )
@@ -67,12 +69,17 @@ func (t *TCPTransport) ListenAndAccept() error {
 
 	go t.startAcceptLoop()
 
+	log.Printf("TCP transport listening on %s", t.ListenAddr)
+
 	return nil
 }
 
 func (t *TCPTransport) startAcceptLoop() {
 	for {
 		conn, err := t.listener.Accept()
+		if errors.Is(err, net.ErrClosed) {
+			return
+		}
 		if err != nil {
 			fmt.Printf("TCP accept error: %v\n", err)
 			continue
@@ -123,3 +130,9 @@ func (t *TCPTransport) handleConn(conn net.Conn) {
 	}
 
 }
+
+// Close implements the Transport interface.
+func (t *TCPTransport) Close() error {
+	close(t.rpcch)
+	return t.listener.Close()
+}

p2p/transport.go

@@ -12,4 +12,5 @@ type Peer interface {
 type Transport interface {
 	ListenAndAccept() error
 	Consume() <-chan RPC
+	Close() error
 }

storage.go

@@ -111,6 +111,10 @@ func (s *Store) Delete(key string) error {
 	return os.RemoveAll(firstPathWithRoot)
 }
 
+func (s *Store) Write(key string, r io.Reader) error {
+	return s.writeStream(key, r)
+}
+
 func (s *Store) Read(key string) (io.Reader, error) {
 	f, err := s.readStream(key)
 	if err != nil {