contrib/raftexample/raft.go
func (rc *raftNode) publishEntries(ents []raftpb.Entry) bool {
	for i := range ents {
		switch ents[i].Type {
		case raftpb.EntryNormal:
			if len(ents[i].Data) == 0 {
				// ignore empty messages
				break
			}
			s := string(ents[i].Data)
			select {
			case rc.commitC <- &s:
			case <-rc.stopc:
				return false
			}

		case raftpb.EntryConfChange:
			var cc raftpb.ConfChange
			cc.Unmarshal(ents[i].Data)
			rc.confState = *rc.node.ApplyConfChange(cc)
			switch cc.Type {
			case raftpb.ConfChangeAddNode:
				if len(cc.Context) > 0 {
					rc.transport.AddPeer(types.ID(cc.NodeID), []string{string(cc.Context)})
				}
			case raftpb.ConfChangeRemoveNode:
				if cc.NodeID == uint64(rc.id) {
					log.Println("I've been removed from the cluster! Shutting down.")
					return false
				}
				rc.transport.RemovePeer(types.ID(cc.NodeID))
			}
		}

		// after commit, update appliedIndex
		rc.appliedIndex = ents[i].Index

		// special nil commit to signal replay has finished
		if ents[i].Index == rc.lastIndex {
			select {
			case rc.commitC <- nil:
			case <-rc.stopc:
				return false
			}
		}
	}
	return true
}
func (rc *raftNode) publishEntries(ents []raftpb.Entry) bool {
	for i := range ents {
		switch ents[i].Type {
		case raftpb.EntryNormal:
			if len(ents[i].Data) == 0 {
				// ignore empty messages
				break
			}
			s := string(ents[i].Data)
			select {
			case rc.commitC <- &s:
			case <-rc.stopc:
				return false
			}

		case raftpb.EntryConfChange:
			var cc raftpb.ConfChange
			cc.Unmarshal(ents[i].Data)
			rc.confState = *rc.node.ApplyConfChange(cc)
			switch cc.Type {
			case raftpb.ConfChangeAddNode:
				if len(cc.Context) > 0 {
					rc.transport.AddPeer(types.ID(cc.NodeID), []string{string(cc.Context)})
				}
			case raftpb.ConfChangeRemoveNode:
				if cc.NodeID == uint64(rc.id) {
					log.Println("I've been removed from the cluster! Shutting down.")
					return false
				}
				rc.transport.RemovePeer(types.ID(cc.NodeID))
			}
		}

		// after commit, update appliedIndex
		rc.appliedIndex = ents[i].Index

		// special nil commit to signal replay has finished
		if ents[i].Index == rc.lastIndex {
			select {
			case rc.commitC <- nil:
			case <-rc.stopc:
				return false
			}
		}
	}
	return true
}
func ( rc * raftNode ) publishEntries ( ents [ ] raftpb . Entry ) bool { for i := range ents { switch ents [ i ] . Type { case raftpb . EntryNormal : if len ( ents [ i ] . Data ) == 0 { // ignore empty messages break } s := string ( ents [ i ] . Data ) select { case rc . commitC <- & s : case <- rc . stopc : return false } case raftpb . EntryConfChange : var cc raftpb . ConfChange cc . Unmarshal ( ents [ i ] . Data ) rc . confState = * rc . node . ApplyConfChange ( cc ) switch cc . Type { case raftpb . ConfChangeAddNode : if len ( cc . Context ) > 0 { rc . transport . AddPeer ( types . ID ( cc . NodeID ) , [ ] string { string ( cc . Context ) } ) } case raftpb . ConfChangeRemoveNode : if cc . NodeID == uint64 ( rc . id ) { log . Println ( " " ) return false } rc . transport . RemovePeer ( types . ID ( cc . NodeID ) ) } } // after commit, update appliedIndex rc . appliedIndex = ents [ i ] . Index // special nil commit to signal replay has finished if ents [ i ] . Index == rc . lastIndex { select { case rc . commitC <- nil : case <- rc . stopc : return false } } } return true }
// publishEntries writes committed log entries to commit channel and returns
// whether all entries could be published.
publishEntries writes committed log entries to commit channel and returns whether all entries could be published .