Bipartite Matching with Apache Hama

BipartiteMatchingCompute.java

    @Override
    public void compute(Iterator<TextPair> messages)
        throws IOException {
      
      if(isMatched()){
        voteToHalt();
        return;
      }

      switch((int)getSuperstepCount() % 4){
        case 0:
          if( Objects.equal(getComponent(), LEFT)){            
            sendMessageToNeighbors(getNewMessage());
          }
          break;

        case 1:
          if( Objects.equal(getComponent(), RIGHT)){
            List<TextPair> buffer = new ArrayList<TextPair>();
            while(messages.hasNext()){
              buffer.add(messages.next());
            }            
            if(buffer.size()>0){
              TextPair luckyMsg = buffer.get(
                  RandomUtils.nextInt(random,buffer.size()));

              Text sourceVertex = getSourceVertex(luckyMsg);
              sendMessage(sourceVertex, getNewMessage());
            }                      
          }
          break;

        case 2:
          if( Objects.equal(getComponent(), LEFT)){
            List<TextPair> buffer = new ArrayList<TextPair>();

            while(messages.hasNext()){
              buffer.add(messages.next());
            }
            if(buffer.size() > 0){
              TextPair luckyMsg = buffer.get(
                  RandomUtils.nextInt(random,buffer.size()));
  
              Text sourceVertex = getSourceVertex(luckyMsg);
              setMatchVertex(sourceVertex);
              sendMessage(sourceVertex, getNewMessage());
            }
          }
          break;

        case 3:
          if(Objects.equal(getComponent(), RIGHT)){
            if(messages.hasNext()){
              Text sourceVertex = getSourceVertex(messages.next());
              setMatchVertex(sourceVertex);
            }
          }
          break;
      }
    }