Transfer: Set max packet size.

udt: Improve error messages. Fix bug in udtSocketSend.processDataMsg, the header was not considered in the max size.
webapi: New /file/view API which is useful for immediate testing of file transfer.

Transfer UDT.go

@@ -44,13 +44,16 @@ func (peer *PeerInfo) startFileTransferUDT(hash []byte, fileSize uint64, offset,
 	virtualConnection.sequenceNumber = sequenceNumber
 	networks.Sequences.RegisterSequenceBi(peer.PublicKey, sequenceNumber, virtualConnection, transferSequenceTimeout, virtualConnection.sequenceTerminate)
 
+	udtConfig := udt.DefaultConfig()
+	udtConfig.MaxPacketSize = protocol.TransferMaxEmbedSize
+
 	// start UDT sender
-	udtConn, err := udt.DialUDT(udt.DefaultConfig(), virtualConnection, virtualConnection.incomingData, virtualConnection.outgoingData, virtualConnection.terminateChan, true)
+	udtConn, err := udt.DialUDT(udtConfig, virtualConnection, virtualConnection.incomingData, virtualConnection.outgoingData, virtualConnection.terminateChan, true)
 	if err != nil {
 		return err
 	}
 
-	defer udtConn.Close() // warning: This is currently blocking in case the other side does not call Close().
+	defer udtConn.Close()
 
 	// Start by sending the header: Total File Size and Transfer Size.
 	header := make([]byte, 16)
@@ -78,8 +81,11 @@ func (peer *PeerInfo) FileTransferRequestUDT(hash []byte, offset, limit uint64)
 	}
 	virtualConnection.sequenceNumber = sequence.SequenceNumber
 
+	udtConfig := udt.DefaultConfig()
+	udtConfig.MaxPacketSize = protocol.TransferMaxEmbedSize
+
 	// start UDT receiver
-	udtListener := udt.ListenUDT(udt.DefaultConfig(), virtualConnection, virtualConnection.incomingData, virtualConnection.outgoingData, virtualConnection.terminateChan)
+	udtListener := udt.ListenUDT(udtConfig, virtualConnection, virtualConnection.incomingData, virtualConnection.outgoingData, virtualConnection.terminateChan)
 
 	// request file transfer
 	peer.sendTransfer(nil, protocol.TransferControlRequestStart, virtualConnection.transferProtocol, hash, offset, limit, virtualConnection.sequenceNumber)

udt/packet/packet.go

@@ -134,7 +134,7 @@ func (h *ctrlHeader) SetHeader(destSockID uint32, ts uint32) {
 func (h *ctrlHeader) writeHdrTo(buf []byte, msgType PacketType, info uint32) (uint, error) {
 	l := len(buf)
 	if l < 16 {
-		return 0, errors.New("packet too small")
+		return 0, errors.New("ctrl packet too small")
 	}
 
 	// Sets the flag bit to indicate this is a control packet
@@ -151,7 +151,7 @@ func (h *ctrlHeader) writeHdrTo(buf []byte, msgType PacketType, info uint32) (ui
 func (h *ctrlHeader) readHdrFrom(data []byte) (addtlInfo uint32, err error) {
 	l := len(data)
 	if l < 16 {
-		return 0, errors.New("packet too small")
+		return 0, errors.New("ctrl packet too small")
 	}
 	addtlInfo = endianness.Uint32(data[4:8])
 	h.ts = endianness.Uint32(data[8:12])

udt/packet/packet_ack.go

@@ -25,7 +25,7 @@ type AckPacket struct {
 func (p *AckPacket) WriteTo(buf []byte) (uint, error) {
 	l := len(buf)
 	if l < 32 {
-		return 0, errors.New("packet too small")
+		return 0, errors.New("ack packet too small")
 	}
 
 	if _, err := p.writeHdrTo(buf, ptAck, p.AckSeqNo); err != nil {
@@ -38,7 +38,7 @@ func (p *AckPacket) WriteTo(buf []byte) (uint, error) {
 	endianness.PutUint32(buf[28:32], p.BuffAvail)
 	if p.IncludeLink {
 		if l < 40 {
-			return 0, errors.New("packet too small")
+			return 0, errors.New("ack packet too small")
 		}
 		endianness.PutUint32(buf[32:36], p.PktRecvRate)
 		endianness.PutUint32(buf[36:40], p.EstLinkCap)
@@ -51,7 +51,7 @@ func (p *AckPacket) WriteTo(buf []byte) (uint, error) {
 func (p *AckPacket) readFrom(data []byte) (err error) {
 	l := len(data)
 	if l < 32 {
-		return errors.New("packet too small")
+		return errors.New("ack packet too small")
 	}
 	if p.AckSeqNo, err = p.readHdrFrom(data); err != nil {
 		return err

udt/packet/packet_data.go

@@ -65,7 +65,7 @@ func (dp *DataPacket) WriteTo(buf []byte) (uint, error) {
 	l := len(buf)
 	ol := 16 + len(dp.Data)
 	if l < ol {
-		return 0, errors.New("packet too small")
+		return 0, errors.New("write data packet too small")
 	}
 	endianness.PutUint32(buf[0:4], dp.Seq.Seq&0x7FFFFFFF)
 	endianness.PutUint32(buf[4:8], dp.msg)
@@ -79,7 +79,7 @@ func (dp *DataPacket) WriteTo(buf []byte) (uint, error) {
 func (dp *DataPacket) readFrom(data []byte) (err error) {
 	l := len(data)
 	if l < 16 {
-		return errors.New("packet too small")
+		return errors.New("read data packet too small")
 	}
 	//dp.seq = endianness.Uint32(data[0:4])
 	dp.msg = endianness.Uint32(data[4:8])

udt/packet/packet_handshake.go

@@ -38,7 +38,7 @@ type HandshakePacket struct {
 func (p *HandshakePacket) WriteTo(buf []byte) (uint, error) {
 	l := len(buf)
 	if l < 64 {
-		return 0, errors.New("packet too small")
+		return 0, errors.New("handshake packet too small")
 	}
 
 	if _, err := p.writeHdrTo(buf, ptHandshake, 0); err != nil {
@@ -64,7 +64,7 @@ func (p *HandshakePacket) WriteTo(buf []byte) (uint, error) {
 func (p *HandshakePacket) readFrom(data []byte) error {
 	l := len(data)
 	if l < 64 {
-		return errors.New("packet too small")
+		return errors.New("handshake packet too small")
 	}
 	if _, err := p.readHdrFrom(data); err != nil {
 		return err

udt/packet/packet_lightack.go

@@ -16,7 +16,7 @@ type LightAckPacket struct {
 func (p *LightAckPacket) WriteTo(buf []byte) (uint, error) {
 	l := len(buf)
 	if l < 20 {
-		return 0, errors.New("packet too small")
+		return 0, errors.New("lightack packet too small")
 	}
 
 	if _, err := p.writeHdrTo(buf, ptAck, 0); err != nil {
@@ -31,7 +31,7 @@ func (p *LightAckPacket) WriteTo(buf []byte) (uint, error) {
 func (p *LightAckPacket) readFrom(data []byte) (err error) {
 	l := len(data)
 	if l < 20 {
-		return errors.New("packet too small")
+		return errors.New("lightack packet too small")
 	}
 	if _, err = p.readHdrFrom(data); err != nil {
 		return err

udt/packet/packet_msgdropreq.go

@@ -18,7 +18,7 @@ type MsgDropReqPacket struct {
 func (p *MsgDropReqPacket) WriteTo(buf []byte) (uint, error) {
 	l := len(buf)
 	if l < 24 {
-		return 0, errors.New("packet too small")
+		return 0, errors.New("drop packet too small")
 	}
 
 	if _, err := p.writeHdrTo(buf, ptMsgDropReq, p.MsgID); err != nil {
@@ -34,7 +34,7 @@ func (p *MsgDropReqPacket) WriteTo(buf []byte) (uint, error) {
 func (p *MsgDropReqPacket) readFrom(data []byte) (err error) {
 	l := len(data)
 	if l < 24 {
-		return errors.New("packet too small")
+		return errors.New("drop packet too small")
 	}
 	if p.MsgID, err = p.readHdrFrom(data); err != nil {
 		return

udt/packet/packet_nak.go

@@ -22,7 +22,7 @@ func (p *NakPacket) WriteTo(buf []byte) (uint, error) {
 
 	l := uint(len(buf))
 	if l < off+uint(4*len(p.CmpLossInfo)) {
-		return 0, errors.New("packet too small")
+		return 0, errors.New("nak packet too small")
 	}
 
 	for _, elm := range p.CmpLossInfo {

udt/packet/packet_userdefctrl.go

@@ -17,7 +17,7 @@ func (p *UserDefControlPacket) WriteTo(buf []byte) (uint, error) {
 	l := len(buf)
 	ol := 16 + len(p.data)
 	if l < ol {
-		return 0, errors.New("packet too small")
+		return 0, errors.New("userdef packet too small")
 	}
 
 	// Sets the flag bit to indicate this is a control packet

udt/udtsocket_send.go

@@ -191,22 +191,17 @@ func (s *udtSocketSend) processDataMsg(isFirst bool, inChan <-chan sendMessage)
 			s.msgSeq++
 		}
 
-		mtu := int(s.socket.maxPacketSize)
+		mtu := int(s.socket.maxPacketSize) - 16
 		msgLen := len(partialSend.content)
 		if msgLen >= mtu {
 			// we are full -- send what we can and leave the rest
-			var dp *packet.DataPacket
+			dp := &packet.DataPacket{
+				Seq:  s.sendPktSeq,
+				Data: partialSend.content[0:mtu],
+			}
 			if msgLen == mtu {
-				dp = &packet.DataPacket{
-					Seq:  s.sendPktSeq,
-					Data: partialSend.content,
-				}
 				s.msgPartialSend = nil
 			} else {
-				dp = &packet.DataPacket{
-					Seq:  s.sendPktSeq,
-					Data: partialSend.content[0:mtu],
-				}
 				s.msgPartialSend = &sendMessage{content: partialSend.content[mtu:], tim: partialSend.tim, ttl: partialSend.ttl}
 			}
 			s.sendPktSeq.Incr()

webapi/API.go

@@ -69,6 +69,7 @@ func Start(ListenAddresses []string, UseSSL bool, CertificateFile, CertificateKe
 	Router.HandleFunc("/warehouse/read", apiWarehouseReadFile).Methods("GET")
 	Router.HandleFunc("/warehouse/delete", apiWarehouseDeleteFile).Methods("GET")
 	Router.HandleFunc("/file/read", apiFileRead).Methods("GET")
+	Router.HandleFunc("/file/view", apiFileView).Methods("GET")
 
 	for _, listen := range ListenAddresses {
 		go startWebServer(listen, UseSSL, CertificateFile, CertificateKey, Router, "API", TimeoutRead, TimeoutWrite)

webapi/File IO.go

@@ -8,7 +8,6 @@ package webapi
 
 import (
 	"errors"
-	"fmt"
 	"io"
 	"net/http"
 	"strconv"
@@ -81,6 +80,7 @@ func apiFileRead(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	// start the reader
 	reader, fileSize, transferSize, err := FileStartReader(peer, fileHash, uint64(offset), uint64(limit))
 	if reader != nil {
 		defer reader.Close()
@@ -89,19 +89,99 @@ func apiFileRead(w http.ResponseWriter, r *http.Request) {
 		w.WriteHeader(http.StatusNotFound)
 		return
 	}
-	fmt.Printf("SUCCESS file size is %d transfer size is %d\n", fileSize, transferSize)
 
-	// Set the Content-Length header, always to the actual size of transferred data.
-	w.Header().Set("Content-Length", strconv.FormatUint(transferSize, 10))
+	// set the right headers
+	setContentLengthRangeHeader(w, uint64(offset), transferSize, fileSize, ranges)
 
-	// Set the Content-Range header if needed.
-	if len(ranges) == 1 {
-		w.Header().Set("Content-Range", "bytes "+strconv.Itoa(offset)+"-"+strconv.Itoa(offset+int(transferSize)-1)+"/"+strconv.FormatUint(fileSize, 10))
-		w.WriteHeader(http.StatusPartialContent)
-	} else {
-		w.WriteHeader(http.StatusOK)
+	// Start sending the data!
+	io.Copy(w, io.LimitReader(reader, int64(transferSize)))
+}
+
+/*
+apiFileView is similar to /file/read but but provides a format parameter. It set the Content-Type header and may manipulate the content on the fly.
+This endpoint supports the Range, Content-Range and Content-Length headers. Multipart ranges are not supported and result in HTTP 400.
+Instead of providing the node ID, the peer ID is also accepted in the &node= parameter.
+The default timeout for connecting to the peer is 10 seconds.
+Formats: 14 = Video
+
+Request:    GET /file/view?hash=[hash]&node=[node ID]&format=[format]
+            Optional: &offset=[offset]&limit=[limit] or via Range header.
+            Optional: &timeout=[seconds]
+Response:   200 with the content
+            206 with partial content
+            400 if the parameters are invalid
+            404 if the file was not found or other error on transfer initiate
+            502 if unable to find or connect to the remote peer in time
+*/
+func apiFileView(w http.ResponseWriter, r *http.Request) {
+	r.ParseForm()
+	var err error
+
+	// validate hashes (must be blake3) and other input
+	fileHash, valid1 := DecodeBlake3Hash(r.Form.Get("hash"))
+	nodeID, valid2 := DecodeBlake3Hash(r.Form.Get("node"))
+	publicKey, err3 := core.PublicKeyFromPeerID(r.Form.Get("node"))
+	if !valid1 || (!valid2 && err3 != nil) {
+		http.Error(w, "", http.StatusBadRequest)
+		return
 	}
 
+	timeoutSeconds, _ := strconv.Atoi(r.Form.Get("timeout"))
+	if timeoutSeconds == 0 {
+		timeoutSeconds = 10
+	}
+	timeout := time.Duration(timeoutSeconds) * time.Second
+
+	offset, _ := strconv.Atoi(r.Form.Get("offset"))
+	limit, _ := strconv.Atoi(r.Form.Get("limit"))
+	format, _ := strconv.Atoi(r.Form.Get("format"))
+
+	// Range header?
+	var ranges []HTTPRange
+	if ranges, err = ParseRangeHeader(r.Header.Get("Range"), -1, true); err != nil || len(ranges) > 1 {
+		http.Error(w, "", http.StatusBadRequest)
+		return
+	} else if len(ranges) == 1 {
+		if ranges[0].length != -1 { // if length is not specified, limit remains 0 which is maximum
+			limit = ranges[0].length
+		}
+		offset = ranges[0].start
+	}
+
+	// try connecting via node ID or peer ID?
+	var peer *core.PeerInfo
+
+	if valid2 {
+		peer, err = PeerConnectNode(nodeID, timeout)
+	} else if err3 == nil {
+		peer, err = PeerConnectPublicKey(publicKey, timeout)
+	}
+	if err != nil {
+		w.WriteHeader(http.StatusBadGateway)
+		return
+	}
+
+	// start the reader
+	reader, fileSize, transferSize, err := FileStartReader(peer, fileHash, uint64(offset), uint64(limit))
+	if reader != nil {
+		defer reader.Close()
+	}
+	if err != nil || reader == nil {
+		w.WriteHeader(http.StatusNotFound)
+		return
+	}
+
+	w.Header().Set("Accept-Ranges", "bytes") // always indicate accepting of Range header
+
+	switch format {
+	case 14:
+		// Video: Indicate MP4 always. There are tons of other MIME types that could be used.
+		w.Header().Set("Content-Type", "video/mp4")
+	}
+
+	// set the right headers
+	setContentLengthRangeHeader(w, uint64(offset), transferSize, fileSize, ranges)
+
 	// Start sending the data!
 	io.Copy(w, io.LimitReader(reader, int64(transferSize)))
 }

webapi/HTTP Range.go

@@ -8,6 +8,7 @@ package webapi
 
 import (
 	"errors"
+	"net/http"
 	"net/textproto"
 	"strconv"
 	"strings"
@@ -92,3 +93,17 @@ func ParseRangeHeader(s string, size int, noMultiRange bool) ([]HTTPRange, error
 	}
 	return ranges, nil
 }
+
+// setContentLengthRangeHeader sets the appropriate Content-Length and Content-Range headers
+func setContentLengthRangeHeader(w http.ResponseWriter, offset, transferSize, fileSize uint64, ranges []HTTPRange) {
+	// Set the Content-Length header, always to the actual size of transferred data.
+	w.Header().Set("Content-Length", strconv.FormatUint(transferSize, 10))
+
+	// Set the Content-Range header if needed.
+	if len(ranges) == 1 {
+		w.Header().Set("Content-Range", "bytes "+strconv.FormatUint(offset, 10)+"-"+strconv.FormatUint(offset+transferSize-1, 10)+"/"+strconv.FormatUint(fileSize, 10))
+		w.WriteHeader(http.StatusPartialContent)
+	} else {
+		w.WriteHeader(http.StatusOK)
+	}
+}