opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/messages/AppendEntries.java

   1 /*
   2  * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8
   9 package org.opendaylight.controller.cluster.raft.messages;
  10
  11 import com.google.protobuf.GeneratedMessage;
  12 import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry;
  13 import org.opendaylight.controller.cluster.raft.ReplicatedLogImplEntry;
  14 import org.opendaylight.controller.cluster.raft.protobuff.client.messages.Payload;
  15 import org.opendaylight.controller.cluster.raft.protobuff.messages.AppendEntriesMessages;
  16
  17 import java.util.ArrayList;
  18 import java.util.Iterator;
  19 import java.util.List;
  20 import java.util.Map;
  21
  22 /**
  23  * Invoked by leader to replicate log entries (§5.3); also used as
  24  * heartbeat (§5.2).
  25  */
  26 public class AppendEntries extends AbstractRaftRPC {
  27
  28     public static final Class SERIALIZABLE_CLASS = AppendEntriesMessages.AppendEntries.class;
  29
  30     private static final org.slf4j.Logger LOG = org.slf4j.LoggerFactory.getLogger(AppendEntries.class);
  31
  32     // So that follower can redirect clients
  33     private final String leaderId;
  34
  35     // Index of log entry immediately preceding new ones
  36     private final long prevLogIndex;
  37
  38     // term of prevLogIndex entry
  39     private final long prevLogTerm;
  40
  41     // log entries to store (empty for heartbeat;
  42     // may send more than one for efficiency)
  43     private final List<ReplicatedLogEntry> entries;
  44
  45     // leader's commitIndex
  46     private final long leaderCommit;
  47
  48     public AppendEntries(long term, String leaderId, long prevLogIndex,
  49         long prevLogTerm, List<ReplicatedLogEntry> entries, long leaderCommit) {
  50         super(term);
  51         this.leaderId = leaderId;
  52         this.prevLogIndex = prevLogIndex;
  53         this.prevLogTerm = prevLogTerm;
  54         this.entries = entries;
  55         this.leaderCommit = leaderCommit;
  56     }
  57
  58     public String getLeaderId() {
  59         return leaderId;
  60     }
  61
  62     public long getPrevLogIndex() {
  63         return prevLogIndex;
  64     }
  65
  66     public long getPrevLogTerm() {
  67         return prevLogTerm;
  68     }
  69
  70     public List<ReplicatedLogEntry> getEntries() {
  71         return entries;
  72     }
  73
  74     public long getLeaderCommit() {
  75         return leaderCommit;
  76     }
  77
  78     @Override public String toString() {
  79         final StringBuilder sb =
  80             new StringBuilder("AppendEntries{");
  81         sb.append("term=").append(getTerm());
  82         sb.append("leaderId='").append(leaderId).append('\'');
  83         sb.append(", prevLogIndex=").append(prevLogIndex);
  84         sb.append(", prevLogTerm=").append(prevLogTerm);
  85         sb.append(", entries=").append(entries);
  86         sb.append(", leaderCommit=").append(leaderCommit);
  87         sb.append('}');
  88         return sb.toString();
  89     }
  90
  91     public <T extends Object> Object toSerializable(){
  92         AppendEntriesMessages.AppendEntries.Builder to = AppendEntriesMessages.AppendEntries.newBuilder();
  93         to.setTerm(this.getTerm())
  94             .setLeaderId(this.getLeaderId())
  95             .setPrevLogTerm(this.getPrevLogTerm())
  96             .setPrevLogIndex(this.getPrevLogIndex())
  97             .setLeaderCommit(this.getLeaderCommit());
  98
  99         for (ReplicatedLogEntry logEntry : this.getEntries()) {
 100
 101             AppendEntriesMessages.AppendEntries.ReplicatedLogEntry.Builder arBuilder =
 102                 AppendEntriesMessages.AppendEntries.ReplicatedLogEntry.newBuilder();
 103
 104             AppendEntriesMessages.AppendEntries.ReplicatedLogEntry.Payload.Builder arpBuilder =
 105                 AppendEntriesMessages.AppendEntries.ReplicatedLogEntry.Payload.newBuilder();
 106
 107             //get the client specific payload extensions and add them to the payload builder
 108             Map<GeneratedMessage.GeneratedExtension, T> map = logEntry.getData().encode();
 109             Iterator<Map.Entry<GeneratedMessage.GeneratedExtension, T>> iter = map.entrySet().iterator();
 110
 111             while (iter.hasNext()) {
 112                 Map.Entry<GeneratedMessage.GeneratedExtension, T> entry = iter.next();
 113                 arpBuilder.setExtension(entry.getKey(), entry.getValue());
 114             }
 115
 116             arpBuilder.setClientPayloadClassName(logEntry.getData().getClientPayloadClassName());
 117
 118             arBuilder.setData(arpBuilder).setIndex(logEntry.getIndex()).setTerm(logEntry.getTerm());
 119             to.addLogEntries(arBuilder);
 120         }
 121
 122         return to.build();
 123     }
 124
 125     public static AppendEntries fromSerializable(Object o){
 126         AppendEntriesMessages.AppendEntries from = (AppendEntriesMessages.AppendEntries) o;
 127
 128         List<ReplicatedLogEntry> logEntryList = new ArrayList<>();
 129         for (AppendEntriesMessages.AppendEntries.ReplicatedLogEntry leProtoBuff : from.getLogEntriesList()) {
 130
 131             Payload payload = null ;
 132             try {
 133                 if(leProtoBuff.getData() != null && leProtoBuff.getData().getClientPayloadClassName() != null) {
 134                     String clientPayloadClassName = leProtoBuff.getData().getClientPayloadClassName();
 135                     payload = (Payload)Class.forName(clientPayloadClassName).newInstance();
 136                     payload = payload.decode(leProtoBuff.getData());
 137                     payload.setClientPayloadClassName(clientPayloadClassName);
 138                 } else {
 139                     LOG.error("Payload is null or payload does not have client payload class name");
 140                 }
 141
 142             } catch (InstantiationException e) {
 143                 LOG.error("InstantiationException when instantiating "+leProtoBuff.getData().getClientPayloadClassName(), e);
 144             } catch (IllegalAccessException e) {
 145                 LOG.error("IllegalAccessException when accessing "+leProtoBuff.getData().getClientPayloadClassName(), e);
 146             } catch (ClassNotFoundException e) {
 147                 LOG.error("ClassNotFoundException when loading "+leProtoBuff.getData().getClientPayloadClassName(), e);
 148             }
 149             ReplicatedLogEntry logEntry = new ReplicatedLogImplEntry(
 150                 leProtoBuff.getIndex(), leProtoBuff.getTerm(), payload);
 151             logEntryList.add(logEntry);
 152         }
 153
 154         AppendEntries to = new AppendEntries(from.getTerm(),
 155             from.getLeaderId(),
 156             from.getPrevLogIndex(),
 157             from.getPrevLogTerm(),
 158             logEntryList,
 159             from.getLeaderCommit());
 160
 161         return to;
 162     }
 163 }