Implement blob manager

Author: ben221199

blob/exchange.go

@@ -132,38 +132,3 @@ func DownloadBlob(ip net.IP, tcpPort int, blobHash string) ([]byte, error) {
 
 	return blobData, nil
 }
-
-// findJSONEnd finds the index of the closing '}' that ends the JSON object.
-// Handles nested braces.
-func findJSONEnd(data []byte) int {
-	depth := 0
-	inString := false
-	escaped := false
-	for i, b := range data {
-		if escaped {
-			escaped = false
-			continue
-		}
-		if b == '\\' && inString {
-			escaped = true
-			continue
-		}
-		if b == '"' {
-			inString = !inString
-			continue
-		}
-		if inString {
-			continue
-		}
-		if b == '{' {
-			depth++
-		}
-		if b == '}' {
-			depth--
-			if depth == 0 {
-				return i
-			}
-		}
-	}
-	return -1
-}

blob/manager.go

@@ -0,0 +1,33 @@
+package blob
+
+import "fmt"
+
+type BlobManager struct {
+	Blobs map[string][]byte
+}
+
+func (blobManager *BlobManager) Has(blobHash string) bool {
+	_, ok := blobManager.Blobs[blobHash]
+	return ok
+}
+
+func (blobManager *BlobManager) Get(blobHash string) []byte {
+	blobData, ok := blobManager.Blobs[blobHash]
+	if ok {
+		return blobData
+	}
+	return nil
+}
+
+func (blobManager *BlobManager) Set(blobHash string, blobData []byte, isStreamDescriptor bool) error {
+	if isStreamDescriptor {
+		// TODO Process SD blob data
+		blobManager.Blobs[blobHash] = blobData
+		fmt.Printf("SD BLOB (%s) = %+v\n", blobHash, string(blobData))
+		return nil
+	}
+	// TODO Process blob data
+	blobManager.Blobs[blobHash] = blobData
+	fmt.Printf("BLOB (%s) = %+v\n", blobHash, blobData)
+	return nil
+}

main.go

@@ -1,6 +1,7 @@
 package main
 
 import "fmt"
+import "lbry/daemon/blob"
 import "lbry/daemon/dht"
 import "lbry/daemon/peer"
 import "lbry/daemon/stream"
@@ -13,6 +14,7 @@ import "sync"
 var wg sync.WaitGroup
 
 func main() {
+	blobManager := blob.BlobManager{}
 	node, _ := dht.NewNode(4444)
 
 	rpcServer := rpc.CreateServer()
@@ -51,7 +53,7 @@ func main() {
 			fmt.Println("Error when getting TCP listener.")
 		}
 		defer listener.Close()
-		reflector.StartServer(listener)
+		reflector.StartServer(blobManager, listener)
 	})
 
 	wg.Go(func() {
@@ -61,7 +63,7 @@ func main() {
 			fmt.Println("Error when getting TCP listener.")
 		}
 		defer listener.Close()
-		peer.StartServer(listener)
+		peer.StartServer(blobManager, listener)
 	})
 
 	wg.Wait()

peer/server.go

@@ -3,21 +3,22 @@ package peer
 import "encoding/json"
 import "fmt"
 import "io"
+import "lbry/daemon/blob"
 import "net"
 import "time"
 
-func StartServer(listener net.Listener) {
+func StartServer(blobManager blob.BlobManager, listener net.Listener) {
 	for {
 		conn, err := listener.Accept()
 		if err != nil {
 			fmt.Println("Error accepting:", err)
 			continue
 		}
-		go handleConnection(conn)
+		go handleConnection(conn, blobManager)
 	}
 }
 
-func handleConnection(conn net.Conn) {
+func handleConnection(conn net.Conn, blobManager blob.BlobManager) {
 	defer conn.Close()
 	conn.SetReadDeadline(time.Now().Add(10 * time.Second)) // Prevent hanging
 
@@ -40,7 +41,7 @@ func handleConnection(conn net.Conn) {
 
 		requestedBlobsValue, hasRequestedBlobs := data["requested_blobs"]
 		if hasRequestedBlobs {
-			responseData["available_blobs"] = getAvailableBlobs(requestedBlobsValue.([]string))
+			responseData["available_blobs"] = getAvailableBlobs(blobManager, requestedBlobsValue.([]string))
 		}
 
 		blobDataPaymentRateValue, hasBlobDataPaymentRate := data["blob_data_payment_rate"]
@@ -53,7 +54,7 @@ func handleConnection(conn net.Conn) {
 
 		requestedBlobValue, hasRequestedBlob := data["requested_blob"]
 		if hasRequestedBlob {
-			incomingBlob, blobData = getRequestedBlob(requestedBlobValue.(string))
+			incomingBlob, blobData = getRequestedBlob(blobManager, requestedBlobValue.(string))
 			responseData["incoming_blob"] = incomingBlob
 		}
 
@@ -64,17 +65,35 @@ func handleConnection(conn net.Conn) {
 	}
 }
 
-func getAvailableBlobs(requestedBlobs []string) []string {
-	// TODO
-	return []string{}
+func getAvailableBlobs(blobManager blob.BlobManager, requestedBlobs []string) []string {
+	var availableBlobs []string
+
+	for _, requestedBlob := range requestedBlobs {
+		if blobManager.Has(requestedBlob) {
+			availableBlobs = append(availableBlobs, requestedBlob)
+		}
+	}
+
+	return availableBlobs
 }
 
 func getBlobDataPaymentRate(blobDataPaymentRate float64) string {
 	// TODO
 	return "RATE_UNSET"
 }
 
-func getRequestedBlob(requestedBlob string) (map[string]any, []byte) {
-	// TODO
-	return map[string]any{}, []byte{}
+func getRequestedBlob(blobManager blob.BlobManager, requestedBlob string) (map[string]any, []byte) {
+	if blobManager.Has(requestedBlob) {
+		blobData := blobManager.Get(requestedBlob)
+		return map[string]any{
+			"blob_hash": requestedBlob,
+			"length":    len(blobData),
+		}, blobData
+	}
+
+	return map[string]any{
+		"blob_hash": "",
+		"length":    0,
+		"error":     "Blob not found",
+	}, nil
 }

reflector/server.go

@@ -3,21 +3,22 @@ package reflector
 import "encoding/json"
 import "fmt"
 import "io"
+import "lbry/daemon/blob"
 import "net"
 import "time"
 
-func StartServer(listener net.Listener) {
+func StartServer(blobManager blob.BlobManager, listener net.Listener) {
 	for {
 		conn, err := listener.Accept()
 		if err != nil {
 			fmt.Println("Error accepting:", err)
 			continue
 		}
-		go handleConnection(conn)
+		go handleConnection(conn, blobManager)
 	}
 }
 
-func handleConnection(conn net.Conn) {
+func handleConnection(conn net.Conn, blobManager blob.BlobManager) {
 	defer conn.Close()
 	conn.SetReadDeadline(time.Now().Add(10 * time.Second)) // Prevent hanging
 
@@ -72,9 +73,12 @@ func handleConnection(conn net.Conn) {
 
 			blobData := make([]byte, blobSize)
 			_, err := io.ReadFull(conn, blobData)
+			if err != nil {
+				conn.Close()
+				return
+			}
 
-			// TODO Process blob data
-			fmt.Printf("BLOB [%d] (%s) = %+v\n", blobSize, blobHash, blobData)
+			err = blobManager.Set(blobHash, blobData, false)
 
 			jsonEncoder.Encode(map[string]any{
 				"received_blob": err == nil,
@@ -101,9 +105,12 @@ func handleConnection(conn net.Conn) {
 
 			sdBlobData := make([]byte, sdBlobSize)
 			_, err := io.ReadFull(conn, sdBlobData)
+			if err != nil {
+				conn.Close()
+				return
+			}
 
-			// TODO Process SD blob data
-			fmt.Printf("SD BLOB [%d] (%s) = %+v\n", sdBlobSize, sdBlobHash, string(sdBlobData))
+			err = blobManager.Set(sdBlobHash, sdBlobData, true)
 
 			jsonEncoder.Encode(map[string]any{
 				"received_sd_blob": err == nil,