DLC Subsystem. All the stuff below + Test + Add FeePerByte to the lit-af utility.

dlc/contract.go

@@ -7,6 +7,7 @@ import (
 )
 
 const COINTYPE_NOT_SET = ^uint32(0) // Max Uint
+const FEEPERBYTE_NOT_SET = ^uint32(0) // Max Uint
 
 // AddContract starts a new draft contract
 func (mgr *DlcManager) AddContract() (*lnutil.DlcContract, error) {
@@ -244,3 +245,44 @@ func (mgr *DlcManager) SetContractCoinType(cIdx uint64, cointype uint32) error {
 
 	return nil
 }
+
+
+//SetContractFeePerByte sets the fee per byte for a particular contract
+func (mgr *DlcManager) SetContractFeePerByte(cIdx uint64, feeperbyte uint32) error {
+	c, err := mgr.LoadContract(cIdx)
+	if err != nil {
+		return err
+	}
+
+	if c.Status != lnutil.ContractStatusDraft {
+		return fmt.Errorf("You cannot change or set the coin type unless" +
+			" the contract is in Draft state")
+	}
+
+	c.FeePerByte = feeperbyte
+
+	mgr.SaveContract(c)
+
+	return nil
+}
+
+
+// //GetContractDivision
+// GetContractDivision(args.CIdx, args.OracleValue)
+// func (mgr *DlcManager) GetContractDivision(cIdx uint64, feeperbyte uint32) error {
+// 	c, err := mgr.LoadContract(cIdx)
+// 	if err != nil {
+// 		return err
+// 	}
+
+// 	if c.Status != lnutil.ContractStatusDraft {
+// 		return fmt.Errorf("You cannot change or set the coin type unless" +
+// 			" the contract is in Draft state")
+// 	}
+
+// 	c.FeePerByte = feeperbyte
+
+// 	mgr.SaveContract(c)
+
+// 	return nil
+// }

litrpc/dlccmds.go

@@ -2,6 +2,7 @@ package litrpc
 
 import (
 	"encoding/hex"
+	"fmt"
 
 	"github.com/mit-dci/lit/dlc"
 	"github.com/mit-dci/lit/lnutil"
@@ -314,6 +315,71 @@ func (r *LitRPC) SetContractCoinType(args SetContractCoinTypeArgs,
 	return nil
 }
 
+
+type SetContractFeePerByteArgs struct {
+	CIdx     uint64
+	FeePerByte uint32
+}
+
+type SetContractFeePerByteReply struct {
+	Success bool
+}
+
+// SetContractFeePerByte sets the coin type the contract will be in. Note that a
+// peer that doesn't have a wallet of that type will automatically decline the
+// contract.
+func (r *LitRPC) SetContractFeePerByte(args SetContractFeePerByteArgs,
+	reply *SetContractFeePerByteReply) error {
+	var err error
+
+	err = r.Node.DlcManager.SetContractFeePerByte(args.CIdx, args.FeePerByte)
+	if err != nil {
+		return err
+	}
+
+	reply.Success = true
+	return nil
+}
+
+//----------------------------------------------------------
+
+type GetContractDivisionArgs struct {
+	CIdx     uint64
+	OracleValue int64
+}
+
+type GetContractDivisionReply struct {
+	ValueOurs int64
+}
+
+// GetContractDivision
+func (r *LitRPC) GetContractDivision(args GetContractDivisionArgs,
+	reply *GetContractDivisionReply) error {
+
+	//err = r.Node.DlcManager.GetContractDivision(args.CIdx, args.OracleValue)
+
+	c, err1 := r.Node.DlcManager.LoadContract(args.CIdx)
+	if err1 != nil {
+		fmt.Errorf("GetContractDivision(): LoadContract err %s\n", err1.Error())
+		return err1
+	}
+
+
+	d, err2 := c.GetDivision(args.OracleValue)
+	if err2 != nil {
+		fmt.Errorf("GetContractDivision(): c.GetDivision err %s\n", err2.Error())
+		return err2
+	}
+	reply.ValueOurs = d.ValueOurs
+
+	return nil
+}
+
+
+//-----------------------------------------------------------
+
+
+
 type OfferContractArgs struct {
 	CIdx    uint64
 	PeerIdx uint32

litrpc/lndcrpcclient.go

@@ -28,9 +28,11 @@ type LndcRpcClient struct {
 	requestNonce       uint64
 	requestNonceMtx    sync.Mutex
 	responseChannelMtx sync.Mutex
-	responseChannels   map[uint64]chan lnutil.RemoteControlRpcResponseMsg
+	responseChannels   map[uint64]chan *lnutil.RemoteControlRpcResponseMsg
 	key                *koblitz.PrivateKey
 	conMtx             sync.Mutex
+
+	chunksOfMsg map[int64]*lnutil.ChunkMsg
 }
 
 // LndcRpcCanConnectLocally checks if we can connect to lit using the normal
@@ -116,9 +118,12 @@ func NewLndcRpcClient(address string, key *koblitz.PrivateKey) (*LndcRpcClient,
 	var err error
 
 	cli := new(LndcRpcClient)
+
+	cli.chunksOfMsg = make(map[int64]*lnutil.ChunkMsg)
+
 	// Create a map of chan objects to receive returned responses on. These channels
 	// are sent to from the ReceiveLoop, and awaited in the Call method.
-	cli.responseChannels = make(map[uint64]chan lnutil.RemoteControlRpcResponseMsg)
+	cli.responseChannels = make(map[uint64]chan *lnutil.RemoteControlRpcResponseMsg)
 
 	//Parse the address we're connecting to
 	who, where := lnutil.ParseAdrString(address)
@@ -158,7 +163,7 @@ func (cli *LndcRpcClient) Call(serviceMethod string, args interface{}, reply int
 
 	// Create the channel to receive the reply on
 	cli.responseChannelMtx.Lock()
-	cli.responseChannels[nonce] = make(chan lnutil.RemoteControlRpcResponseMsg)
+	cli.responseChannels[nonce] = make(chan *lnutil.RemoteControlRpcResponseMsg)
 	cli.responseChannelMtx.Unlock()
 
 	// Send the message in a goroutine
@@ -221,6 +226,35 @@ func (cli *LndcRpcClient) ReceiveLoop() {
 			return
 		}
 		msg = msg[:n]
+
+		if msg[0] == lnutil.MSGID_CHUNKS_BEGIN {
+
+			beginChunksMsg, _ := lnutil.NewChunksBeginMsgFromBytes(msg, 0)
+
+			msg_tmp := new(lnutil.ChunkMsg)
+			msg_tmp.TimeStamp = beginChunksMsg.TimeStamp
+			cli.chunksOfMsg[beginChunksMsg.TimeStamp] = msg_tmp
+
+			continue
+		}
+
+		if msg[0] == lnutil.MSGID_CHUNK_BODY {
+
+			chunkMsg, _ := lnutil.NewChunkMsgFromBytes(msg, 0)
+			cli.chunksOfMsg[chunkMsg.TimeStamp].Data = append(cli.chunksOfMsg[chunkMsg.TimeStamp].Data, chunkMsg.Data...)
+
+			continue
+		}
+
+		if msg[0] == lnutil.MSGID_CHUNKS_END {
+
+			endChunksMsg, _ := lnutil.NewChunksBeginMsgFromBytes(msg, 0)
+			msg = cli.chunksOfMsg[endChunksMsg.TimeStamp].Data
+
+		}		
+
+
+
 		// We only care about RPC responses (for now)
 		if msg[0] == lnutil.MSGID_REMOTE_RPCRESPONSE {
 			// Parse the received message
@@ -239,7 +273,7 @@ func (cli *LndcRpcClient) ReceiveLoop() {
 				// reply and therefore, it could have not blocked and just
 				// ignore the return value.
 				select {
-				case responseChan <- response:
+				case responseChan <- &response:
 				default:
 				}
 

lndc/conn.go

@@ -7,6 +7,7 @@ import (
 	"math"
 	"net"
 	"time"
+	"encoding/binary"
 
 	"github.com/mit-dci/lit/crypto/koblitz"
 	"github.com/mit-dci/lit/lnutil"
@@ -151,27 +152,62 @@ func (c *Conn) Write(b []byte) (n int, err error) {
 
 	// If we need to split the message into fragments, then we'll write
 	// chunks which maximize usage of the available payload.
-	chunkSize := math.MaxUint16
+	chunkSize := math.MaxUint16 - 1000
+
+	lenb := int(len(b))
+	curts := (time.Now().UnixNano())
+
+	beginChunksMsgBodyBytes := new(bytes.Buffer)
+	beginChunksMsgBodyBytes.WriteByte(lnutil.MSGID_CHUNKS_BEGIN)
+	binary.Write(beginChunksMsgBodyBytes, binary.BigEndian, curts)
+
+	// Start saving chunks
+	chunk_err := c.noise.WriteMessage(c.conn, beginChunksMsgBodyBytes.Bytes())
+	if chunk_err != nil {
+		return 0, chunk_err
+	}
 
-	bytesToWrite := len(b)
 	bytesWritten := 0
+	bytesToWrite := lenb
+
 	for bytesWritten < bytesToWrite {
 		// If we're on the last chunk, then truncate the chunk size as
 		// necessary to avoid an out-of-bounds array memory access.
 		if bytesWritten+chunkSize > len(b) {
-			chunkSize = len(b) - bytesWritten
+			chunkSize = lenb - bytesWritten
 		}
 
 		// Slice off the next chunk to be written based on our running
 		// counter and next chunk size.
 		chunk := b[bytesWritten : bytesWritten+chunkSize]
-		if err := c.noise.WriteMessage(c.conn, chunk); err != nil {
-			return bytesWritten, err
+
+		// Wrap chunk in a MSGID_CHUNK_BODY message
+		chunkMsgBodyBytes := new(bytes.Buffer)
+		chunkMsgBodyBytes.WriteByte(lnutil.MSGID_CHUNK_BODY)
+		binary.Write(chunkMsgBodyBytes, binary.BigEndian, curts)
+		binary.Write(chunkMsgBodyBytes, binary.BigEndian, int32(chunkSize))
+		chunkMsgBodyBytes.Write(chunk)
+
+
+		chunk_err = c.noise.WriteMessage(c.conn, chunkMsgBodyBytes.Bytes())
+		if chunk_err != nil {
+			return 0, chunk_err
 		}
 
 		bytesWritten += len(chunk)
 	}
 
+	// Actually send a message (unwrap and send)
+	endChunksMsgBodyBytes := new(bytes.Buffer)
+	endChunksMsgBodyBytes.WriteByte(lnutil.MSGID_CHUNKS_END)
+	binary.Write(endChunksMsgBodyBytes, binary.BigEndian, curts)
+
+	chunk_err = c.noise.WriteMessage(c.conn, endChunksMsgBodyBytes.Bytes())
+	if chunk_err != nil {
+		return 0, chunk_err
+	}
+
+
 	return bytesWritten, nil
 }
 

lnp2p/msgproc.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"github.com/mit-dci/lit/logging"
 	"sync"
+	"github.com/mit-dci/lit/lnutil"
 )
 
 // ParseFuncType is the type of a Message parser function.
@@ -28,6 +29,8 @@ type MessageProcessor struct {
 	// TODO Evaluate if this mutex is even necessary?
 	active bool
 	actmtx *sync.Mutex
+
+	ChunksOfMsg map[int64]*lnutil.ChunkMsg
 }
 
 // NewMessageProcessor processes messages coming in from over the network.
@@ -36,6 +39,7 @@ func NewMessageProcessor() MessageProcessor {
 		handlers: [256]*messagehandler{},
 		active:   false,
 		actmtx:   &sync.Mutex{},
+		ChunksOfMsg: make(map[int64]*lnutil.ChunkMsg),
 	}
 }
 
@@ -77,6 +81,43 @@ func (mp *MessageProcessor) HandleMessage(peer *Peer, buf []byte) error {
 
 	// First see if we have handlers defined for this message type.
 	mtype := buf[0]
+
+	if mtype == 0xB2{
+
+		msg, _ := lnutil.NewChunksBeginMsgFromBytes(buf, peer.GetIdx())
+
+		chunk_msg := new(lnutil.ChunkMsg)
+		chunk_msg.TimeStamp = msg.TimeStamp
+
+		mp.ChunksOfMsg[msg.TimeStamp] = chunk_msg
+
+		return nil
+
+	}
+
+	if mtype == 0xB3{
+
+		msg, _ := lnutil.NewChunkMsgFromBytes(buf, peer.GetIdx())
+		mp.ChunksOfMsg[msg.TimeStamp].Data = append(mp.ChunksOfMsg[msg.TimeStamp].Data, msg.Data...)
+
+		return nil
+
+	}
+
+	if mtype == 0xB4{
+
+		msg, _ := lnutil.NewChunksEndMsgFromBytes(buf, peer.GetIdx())
+
+		buf = mp.ChunksOfMsg[msg.TimeStamp].Data
+		mtype = buf[0]
+
+		delete(mp.ChunksOfMsg, msg.TimeStamp)
+
+	}
+
+
+
+
 	h := mp.handlers[mtype]
 	if h == nil {
 		return fmt.Errorf("no handler found for messasge of type %x", mtype)

lnp2p/peermgr.go

@@ -7,6 +7,7 @@ import (
 	"net"
 	"sync"
 	"time"
+	"math"
 
 	"github.com/mit-dci/lit/btcutil/hdkeychain"
 	"github.com/mit-dci/lit/crypto/koblitz"
@@ -87,6 +88,25 @@ func NewPeerManager(rootkey *hdkeychain.ExtendedKey, pdb lncore.LitPeerStorage,
 		mtx:            &sync.Mutex{},
 	}
 
+
+	// Clear ChunksOfMsg in case of incomplete chunks transmittion.
+	// Try to clean the map every 5 minutes. Therefore message have to
+	// be transmitted within a 5 minutes.
+	go func(){
+
+		for {
+
+			time.Sleep(5 * time.Minute)
+
+			for k := range pm.mproc.ChunksOfMsg {
+				tdelta := time.Now().UnixNano() - k
+				if tdelta > 3*int64(math.Pow10(11)) {
+					delete(pm.mproc.ChunksOfMsg, k)
+				}
+			}
+		}
+	}()
+
 	return pm, nil
 }