hyperledger-archives · srderson · Jun 23, 2016 · Jun 22, 2016 · Jun 22, 2016 · Jun 22, 2016
diff --git a/core/chaincode/chaincode_support.go b/core/chaincode/chaincode_support.go
@@ -20,6 +20,7 @@ import (
 	"bytes"
 	"fmt"
 	"io"
+	"strconv"
 	"sync"
 	"time"
 
@@ -129,6 +130,21 @@ func NewChaincodeSupport(chainname ChainName, getPeerEndpoint func() (*pb.PeerEn
 		s.peerTLSSvrHostOrd = viper.GetString("peer.tls.serverhostoverride")
 	}
 
+	kadef := 0
+	if ka := viper.GetString("chaincode.keepalive"); ka == "" {
+		s.keepalive = time.Duration(kadef) * time.Second
+	} else {
+		t, terr := strconv.Atoi(ka)
+		if terr != nil {
+			chaincodeLogger.Errorf("Invalid keepalive value %s (%s) defaulting to %d", ka, terr, kadef)
+			t = kadef
+		} else if t <= 0 {
+			chaincodeLogger.Debugf("Turn off keepalive(value %s)", ka)
+			t = kadef
+		}
+		s.keepalive = time.Duration(t) * time.Second
+	}
+
 	return s
 }
 
@@ -153,6 +169,7 @@ type ChaincodeSupport struct {
 	peerTLSCertFile      string
 	peerTLSKeyFile       string
 	peerTLSSvrHostOrd    string
+	keepalive            time.Duration
 }
 
 // DuplicateChaincodeHandlerError returned if attempt to register same chaincodeID while a stream already exists.

diff --git a/core/chaincode/chaincodetest.yaml b/core/chaincode/chaincodetest.yaml
@@ -393,6 +393,11 @@ chaincode:
     # the image
     installpath: /opt/gopath/bin/
 
+    #keepalive in seconds. In situations where the communiction goes through a 
+    #proxy that does not support keep-alive, this parameter will maintain connection
+    #between peer and chaincode.
+    #A value <= 0 turns keepalive off
+    keepalive: 1
 ###############################################################################
 #
 #    Ledger section - ledger configuration encompases both the blockchain

diff --git a/core/chaincode/exectransaction_test.go b/core/chaincode/exectransaction_test.go
@@ -469,6 +469,9 @@ func TestExecuteInvokeTransaction(t *testing.T) {
 	//TLS is on by default. This is the ONLY test that does NOT use TLS
 	viper.Set("peer.tls.enabled", false)
 
+	//turn OFF keepalive. All other tests use keepalive
+	viper.Set("peer.chaincode.keepalive", "0")
+
 	if viper.GetBool("peer.tls.enabled") {
 		creds, err := credentials.NewServerTLSFromFile(viper.GetString("peer.tls.cert.file"), viper.GetString("peer.tls.key.file"))
 		if err != nil {

diff --git a/core/chaincode/handler.go b/core/chaincode/handler.go
@@ -264,6 +264,16 @@ func (handler *Handler) triggerNextState(msg *pb.ChaincodeMessage, send bool) {
 	handler.nextState <- &nextStateInfo{msg, send}
 }
 
+func (handler *Handler) waitForKeepaliveTimer() <-chan time.Time {
+	if handler.chaincodeSupport.keepalive > 0 {
+		c := time.After(handler.chaincodeSupport.keepalive)
+		return c
+	}
+	//no one will signal this channel, listner blocks forever
+	c := make(chan time.Time, 1)
+	return c
+}
+
 func (handler *Handler) processStream() error {
 	defer handler.deregister()
 	msgAvail := make(chan *pb.ChaincodeMessage)
@@ -307,6 +317,13 @@ func (handler *Handler) processStream() error {
 
 			// we can spin off another Recv again
 			recv = true
+
+			if in.Type == pb.ChaincodeMessage_KEEPALIVE {
+				chaincodeLogger.Debug("Received KEEPALIVE Response")
+				// Received a keep alive message, we don't do anything with it for now
+				// and it does not touch the state machine
+				continue
+			}
 		case nsInfo = <-handler.nextState:
 			in = nsInfo.msg
 			if in == nil {
@@ -315,12 +332,29 @@ func (handler *Handler) processStream() error {
 				return err
 			}
 			chaincodeLogger.Debugf("[%s]Move state message %s", shortuuid(in.Uuid), in.Type.String())
+		case <-handler.waitForKeepaliveTimer():
+			if handler.chaincodeSupport.keepalive <= 0 {
+				chaincodeLogger.Errorf("Invalid select: keepalive not on (keepalive=%d)", handler.chaincodeSupport.keepalive)
+				continue
+			}
+
+			//TODO we could use this to hook into container lifecycle (kill the chaincode if not in use, etc)
+			kaerr := handler.serialSend(&pb.ChaincodeMessage{Type: pb.ChaincodeMessage_KEEPALIVE})
+			if kaerr != nil {
+				chaincodeLogger.Errorf("Error sending keepalive, err=%s", kaerr)
+			} else {
+				chaincodeLogger.Debug("Sent KEEPALIVE request")
+			}
+			//keepalive message kicked in. just continue
+			continue
 		}
+
 		err = handler.HandleMessage(in)
 		if err != nil {
 			chaincodeLogger.Errorf("[%s]Error handling message, ending stream: %s", shortuuid(in.Uuid), err)
 			return fmt.Errorf("Error handling message, ending stream: %s", err)
 		}
+
 		if nsInfo != nil && nsInfo.sendToCC {
 			chaincodeLogger.Debugf("[%s]sending state message %s", shortuuid(in.Uuid), in.Type.String())
 			if err = handler.serialSend(in); err != nil {

diff --git a/core/chaincode/shim/chaincode.go b/core/chaincode/shim/chaincode.go
@@ -224,8 +224,14 @@ func chatWithPeer(chaincodename string, stream PeerChaincodeStream, cc Chaincode
 				err = fmt.Errorf("Error handling message: %s", err)
 				return
 			}
-			if nsInfo != nil && nsInfo.sendToCC {
-				chaincodeLogger.Debugf("[%s]send state message %s", shortuuid(in.Uuid), in.Type.String())
+
+			//keepalive messages are PONGs to the fabric's PINGs
+			if (nsInfo != nil && nsInfo.sendToCC) || (in.Type == pb.ChaincodeMessage_KEEPALIVE) {
+				if in.Type == pb.ChaincodeMessage_KEEPALIVE {
+					chaincodeLogger.Debug("Sending KEEPALIVE response")
+				} else {
+					chaincodeLogger.Debugf("[%s]send state message %s", shortuuid(in.Uuid), in.Type.String())
+				}
 				if err = handler.serialSend(in); err != nil {
 					err = fmt.Errorf("Error sending %s: %s", in.Type.String(), err)
 					return
@@ -832,6 +838,7 @@ func (stub *ChaincodeStub) insertRowInternal(tableName string, row Row, update b
 }
 
 // ------------- ChaincodeEvent API ----------------------
+
 // SetEvent saves the event to be sent when a transaction is made part of a block
 func (stub *ChaincodeStub) SetEvent(name string, payload []byte) error {
 	stub.chaincodeEvent = &pb.ChaincodeEvent{EventName: name, Payload: payload}

diff --git a/core/chaincode/shim/handler.go b/core/chaincode/shim/handler.go
@@ -866,6 +866,11 @@ func (handler *Handler) handleQueryChaincode(chaincodeName string, function stri
 
 // handleMessage message handles loop for shim side of chaincode/validator stream.
 func (handler *Handler) handleMessage(msg *pb.ChaincodeMessage) error {
+	if msg.Type == pb.ChaincodeMessage_KEEPALIVE {
+		// Received a keep alive message, we don't do anything with it for now
+		// and it does not touch the state machine
+		return nil
+	}
 	chaincodeLogger.Debugf("[%s]Handling ChaincodeMessage of type: %s(state:%s)", shortuuid(msg.Uuid), msg.Type, handler.FSM.Current())
 	if handler.FSM.Cannot(msg.Type.String()) {
 		errStr := fmt.Sprintf("[%s]Chaincode handler FSM cannot handle message (%s) with payload size (%d) while in state: %s", msg.Uuid, msg.Type.String(), len(msg.Payload), handler.FSM.Current())

diff --git a/peer/core.yaml b/peer/core.yaml
@@ -314,6 +314,14 @@ chaincode:
     # the image
     installpath: /opt/gopath/bin/
 
+    # keepalive in seconds. In situations where the communiction goes through a 
+    # proxy that does not support keep-alive, this parameter will maintain connection
+    # between peer and chaincode.
+    # A value <= 0 turns keepalive off
+    keepalive: 0
+
+###############################################################################
+#
 ###############################################################################
 #
 #    Ledger section - ledger configuration encompases both the blockchain

diff --git a/protos/api.pb.go b/protos/api.pb.go
diff --git a/protos/chaincode.pb.go b/protos/chaincode.pb.go
diff --git a/protos/chaincode.proto b/protos/chaincode.proto
@@ -136,6 +136,7 @@ message ChaincodeMessage {
         RANGE_QUERY_STATE = 17;
         RANGE_QUERY_STATE_NEXT = 18;
         RANGE_QUERY_STATE_CLOSE = 19;
+        KEEPALIVE = 20;
     }
 
     Type type = 1;