Drools support for kafka topics
[policy/drools-pdp.git] / feature-pooling-dmaap / src / main / java / org / onap / policy / drools / pooling / message / Leader.java
1 /*
2  * ============LICENSE_START=======================================================
3  * ONAP
4  * ================================================================================
5  * Copyright (C) 2018-2019, 2021 AT&T Intellectual Property. All rights reserved.
6  * ================================================================================
7  * Licensed under the Apache License, Version 2.0 (the "License");
8  * you may not use this file except in compliance with the License.
9  * You may obtain a copy of the License at
10  *
11  *      http://www.apache.org/licenses/LICENSE-2.0
12  *
13  * Unless required by applicable law or agreed to in writing, software
14  * distributed under the License is distributed on an "AS IS" BASIS,
15  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16  * See the License for the specific language governing permissions and
17  * limitations under the License.
18  * ============LICENSE_END=========================================================
19  */
20
21 package org.onap.policy.drools.pooling.message;
22
23 import lombok.NoArgsConstructor;
24 import org.onap.policy.drools.pooling.PoolingFeatureException;
25
26 /**
27  * Indicates that the "source" of this message is now the "lead" host.
28  */
29 @NoArgsConstructor
30 public class Leader extends MessageWithAssignments {
31
32     /**
33      * Constructor.
34      *
35      * @param source host on which the message originated
36      * @param assignments assignments
37      */
38     public Leader(String source, BucketAssignments assignments) {
39         super(source, assignments);
40     }
41
42     /**
43      * Also verifies that buckets have been assigned and that the source is
44      * indeed the leader.
45      */
46     @Override
47     public void checkValidity() throws PoolingFeatureException {
48
49         super.checkValidity();
50
51         BucketAssignments assignments = getAssignments();
52         if (assignments == null) {
53             throw new PoolingFeatureException("missing message bucket assignments");
54         }
55
56         String leader = getSource();
57
58         if (!assignments.hasAssignment(leader)) {
59             throw new PoolingFeatureException("leader " + leader + " has no bucket assignments");
60         }
61
62         for (String host : assignments.getHostArray()) {
63             if (host.compareTo(leader) < 0) {
64                 throw new PoolingFeatureException("invalid leader " + leader + ", should be " + host);
65             }
66         }
67     }
68
69 }