Github user smarthi commented on a diff in the pull request:
https://github.com/apache/incubator-pirk/pull/74#discussion_r76515127
--- Diff:
src/main/java/org/apache/pirk/responder/wideskies/storm/EncColMultBolt.java ---
@@ -0,0 +1,131 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.pirk.responder.wideskies.storm;
+
+import org.apache.storm.task.OutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.topology.base.BaseRichBolt;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Tuple;
+import org.apache.storm.tuple.Values;
+import org.slf4j.LoggerFactory;
+
+import java.math.BigInteger;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Bolt class to perform encrypted column multiplication
+ * <p>
+ * Takes {@code <columnIndex, columnValue>} tuples as input and aggregates
(multiplies) the columnValues for a given columnIndex as they are received.
+ * <p>
+ * EncRowCalcBolts send flush signals to the EncColMultBolts indicating
that they have finished sending all tuples for a session. Whenever a flush
signal is
+ * received from a EncRowCalcBolt, the num of received flush signals is
tallied until each encrypted row has emitted a flush signal (there are
2^hashBitSize
+ * rows).
+ * <p>
+ * Once a flush signal has been received from each row, all {@code
<columnIndex, aggregate colVal product>} tuples are sent to the OutputBolt and
a session_end
+ * signal is sent back to each EncRowMultBolt.
+ * <p>
+ * The EncRowMultBolts buffer their output from the time that they send a
flush signal to the EncColMultBolts until the time that they receive a
session_end
+ * signal from the EncColMultBolts.
+ *
+ */
+public class EncColMultBolt extends BaseRichBolt
+{
+ private static final long serialVersionUID = 1L;
+
+ private static final org.slf4j.Logger logger =
LoggerFactory.getLogger(EncColMultBolt.class);
+
+ private OutputCollector outputCollector;
+
+ private BigInteger nSquared;
+ private long numFlushSignals;
+ private Long totalFlushSignals;
+
+ // This is the main object here. It holds column Id -> aggregated product
+ private HashMap<Long,BigInteger> resultsMap = new
HashMap<Long,BigInteger>();
--- End diff --
use Map<Long,BigInteger> on LHS
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
