4
\$\begingroup\$

I am using below class to send data to our messaging queue by using socket either in a synchronous way or asynchronous way as shown below. It depends on requirement whether I want to call synchronous or asynchronous method to send data on a socket.

  • asynchronous - It sends data asynchronously and we don't block the thread which is sending data. If acknowledgment is not received then it will retry again from the background thread which is started in SendRecord constructor only. This can only be done by calling one method sendToQueueAsync.
  • synchronous - It sends data synchronously on a socket. It internally calls doSendAsync method and then sleep for a particular timeout period and if acknowledgment is not received then it removes from cache bucket so that we don't retry again. There are two ways to send data through sync mean either calling sendToQueueSync method with two parameters or with three parameters.

So the only difference between those two above case is - For async case, I need to retry at all cost if acknowledgment is not received but for sync I don't need to retry at all and that's why I am storing more state in a PendingMessage class.

ResponsePoller is a class which receives the acknowledgment for the data that was sent to our messaging queue and then calls handleAckReceived method below to remove the address so that we don't retry after receiving the acknowledgment.

public class SendRecord {
  private final ScheduledExecutorService executorService = Executors.newScheduledThreadPool(2);
  private final Cache<Long, PendingMessage> cache = CacheBuilder.newBuilder().maximumSize(1000000)
      .concurrencyLevel(100).build();

  private static class Holder {
    private static final SendRecord INSTANCE = new SendRecord();
  }

  public static SendRecord getInstance() {
    return Holder.INSTANCE;
  }

  private SendRecord() {
    executorService.scheduleAtFixedRate(new Runnable() {
      @Override
      public void run() {
        handleRetry();
      }
    }, 0, 1, TimeUnit.SECONDS);
  }

  // this will retry to send data again if acknowledgment is not received
  private void handleRetry() {
    List<PendingMessage> messages = new ArrayList<>(cache.asMap().values());
    for (PendingMessage message : messages) {
      if (message.hasExpired()) {
        if (message.shouldRetry()) {
          message.markResent();
          doSendAsync(message);
        } else {
          cache.invalidate(message.getAddress());
        }
      }
    }
  }

  // called by multiple threads concurrently
  public boolean sendToQueueAsync(final long address, final byte[] encodedRecords) {
    PendingMessage m = new PendingMessage(address, encodedRecords, true);
    cache.put(address, m);
    return doSendAsync(m);
  }

  // called by above method and also by handleRetry method
  private boolean doSendAsync(final PendingMessage pendingMessage) {
    Optional<SocketHolder> liveSocket = SocketManager.getInstance().getNextSocket();
    ZMsg msg = new ZMsg();
    msg.add(pendingMessage.getEncodedRecords());
    try {
      // this returns instantly
      return msg.send(liveSocket.get().getSocket());
    } finally {
      msg.destroy();
    }
  }

  // called by send method below
  private boolean doSendAsync(final PendingMessage pendingMessage, final Socket socket) {
    ZMsg msg = new ZMsg();
    msg.add(pendingMessage.getEncodedRecords());
    try {
      // this returns instantly
      return msg.send(socket);
    } finally {
      msg.destroy();
    }
  }

  // called by multiple threads to send data synchronously without passing socket
  public boolean sendToQueueSync(final long address, final byte[] encodedRecords) {
    PendingMessage m = new PendingMessage(address, encodedRecords, false);
    cache.put(address, m);
    try {
      if (doSendAsync(m)) {
        return m.waitForAck();
      }
      return false;
    } finally {
      cache.invalidate(address);
    }
  }

  // called by a threads to send data synchronously but with socket as the parameter
  public boolean sendToQueueSync(final long address, final byte[] encodedRecords, final Socket socket) {
    PendingMessage m = new PendingMessage(address, encodedRecords, false);
    cache.put(address, m);
    try {
      if (doSendAsync(m, socket)) {
        return m.waitForAck();
      }
      return false;
    } finally {
      cache.invalidate(address);
    }
  }

  public void handleAckReceived(final long address) {
    PendingMessage record = cache.getIfPresent(address);
    if (record != null) {
      record.ackReceived();
      cache.invalidate(address);
    }
  }
}

As of now I am duplicating lot of code in my above class. Callers will only call either of below three methods to send data but just to have below three methods, I have lot of duplication. There might be better way I think to do same above stuff by following oops standard. Maybe we need an interface and then its implementation. Opting for code review to see if there is any better way.

  • sendToQueueAsync by passing two parameters
  • sendToQueueSync by passing two parameters
  • sendToQueueSync by passing three parameters

We can have an interface specifying send() method and one implementation for each sending strategy so total 3 classes in total and then use factory pattern to get actual implementation depending on parameters passed. Not sure if this is the best way.

\$\endgroup\$
  • \$\begingroup\$ If your concern is code duplication in the callers, it would help to see some of the caller code. Also: you have a Cache class, is this Android? If not what library is the Cache from? \$\endgroup\$ – markspace Jan 14 '18 at 5:37
  • \$\begingroup\$ And before I forget: you have a List for tracking messages which may need to be resent. Is there a spec for what happens if the app terminates abruptly? As-is, the messages held in memory will be dumped and never sent. \$\endgroup\$ – markspace Jan 14 '18 at 5:39
  • \$\begingroup\$ This is not android. Here cache is Guava cache. I think that's ok if an app terminates abruptly for now but in future maybe we can write to disk whatever we are holding in cache so whenever app terminates abruptly then it can read those files from disk and send it back whenevr it restarts or maybe if there is any other better way, I am open for that as well. My concern is in this class only not the callers code. \$\endgroup\$ – david Jan 14 '18 at 5:49
3
+25
\$\begingroup\$

I would collapse your doSendAsync methods into one, since they are private. Doing that would help refactor both sendToQueueSync methods so they don't have duplicated code, killing two birds with a stone.

public boolean sendToQueueAsync(final long address, final byte[] encodedRecords) {
    PendingMessage m = new PendingMessage(address, encodedRecords, true);
    cache.put(address, m);
    return doSendAsync(m, null);
}

private boolean doSendAsync(final PendingMessage pendingMessage, final Socket socket) {
    Socket actualSocket = socket;

    if(actualSocket == null){
        Optional<SocketHolder> liveSocket = SocketManager.getInstance().getNextSocket();
        actualSocket = liveSocket.get().getSocket();
    }

    ZMsg msg = new ZMsg();
    msg.add(pendingMessage.getEncodedRecords());
    try {
      return msg.send(actualSocket );
    } finally {
      msg.destroy();
    }
}

public boolean sendToQueueSync(final long address, final byte[] encodedRecords) {
    return sendToQueueSync(address, encodedRecords, null);
}

public boolean sendToQueueSync(final long address, final byte[] encodedRecords, final Socket socket) {
    PendingMessage m = new PendingMessage(address, encodedRecords, false);
    cache.put(address, m);
    try {
      if (doSendAsync(m, socket)) {
        return m.waitForAck();
      }
      return false;
    } finally {
      cache.invalidate(address);
    }
}

Furthermore, you can explore the idea of having a Socket being passed on the constructor. That way you would always have only one method variant on your public API, but I guess that comes with issues with socket life management that you might not want to deal with so it's up to your discretion.

If you consider such option you can further reduce the amount of code needed and you will provide the expected API, usually there's a rule that says that for every asynchronous call there be an equivalent synchronous call (with the same parameters). In your approach you break that rule. Here's how it would look:

public class SendRecord
{
    private final ScheduledExecutorService executorService = Executors.newScheduledThreadPool(2);
    private final Cache<Long, PendingMessage> cache = CacheBuilder.newBuilder().maximumSize(1000000)
        .concurrencyLevel(100).build();
    private Socket socket;

    public SendRecord(){
        this(SocketManager.getInstance().getNextSocket().get.getSocket());
    }

    public SendRecord(Socket socket){
        this.socket = socket;
    }

    private static class Holder
    {
        private static final SendRecord INSTANCE = new SendRecord();
    }

    public static SendRecord getInstance()
    {
        return Holder.INSTANCE;
    }
    private SendRecord()
    {
        executorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run()
            {
                handleRetry();
            }
        }, 0, 1, TimeUnit.SECONDS);
    }

    // this will retry to send data again if acknowledgment is not received
    private void handleRetry()
    {
        List<PendingMessage> messages = new ArrayList<>(cache.asMap().values());
        for (PendingMessage message : messages)
        {
            if (message.hasExpired())
            {
                if (message.shouldRetry())
                {
                    message.markResent();
                    doSendAsync(message);
                }
                else
                {
                    cache.invalidate(message.getAddress());
                }
            }
        }
    }

    // called by multiple threads concurrently
    public boolean sendToQueueAsync(final long address, final byte[] encodedRecords)
    {
        PendingMessage m = new PendingMessage(address, encodedRecords, true);
        cache.put(address, m);
        return doSendAsync(m);
    }

    // called by send method below
    private boolean doSendAsync(final PendingMessage pendingMessage)
    {
        ZMsg msg = new ZMsg();
        msg.add(pendingMessage.getEncodedRecords());
        try
        {
            // this returns instantly
            return msg.send(socket);
        }
        finally
        {
            msg.destroy();
        }
    }
    // called by a threads to send data synchronously but with socket as the parameter
    public boolean sendToQueueSync(final long address, final byte[] encodedRecords)
    {
        PendingMessage m = new PendingMessage(address, encodedRecords, false);
        cache.put(address, m);
        try
        {
            if (doSendAsync(m, socket))
            {
                return m.waitForAck();
            }
            return false;
        }
        finally
        {
            cache.invalidate(address);
        }
    }

    public void handleAckReceived(final long address)
    {
        PendingMessage record = cache.getIfPresent(address);
        if (record != null)
        {
            record.ackReceived();
            cache.invalidate(address);
        }
    }
}

As suggested in a previous question of your (lost track of the specific question) you should also think about implementing a Future. Doing a synchronous call with a future should be as simple as future.get()

\$\endgroup\$
  • \$\begingroup\$ I liked your idea of collapsing doSendAsync methods into one. Thanks for that. Also if you see my handleRetry() method, that is also calling one of doSendAsync method so there will be a change in handleRetry() method as well? So we would be passing like this doSendAsync(message, null); right? Also do you think it's better if we create an interface or base abstract class and then we can have three different implementations for that? And then we can use some factory pattern to get that implementation we need? Let me know your thoughts on this. \$\endgroup\$ – david Jan 16 '18 at 20:50
  • \$\begingroup\$ Also I noticed you mentioned about taking Socket as parameter in the constructor? I didn't quite understand how will that look like with my design? Can you provide an example on that if possible so that I can understand better? \$\endgroup\$ – david Jan 16 '18 at 20:51
  • \$\begingroup\$ @david I think there's no need to create an abstract class. The reason is that you don't really have different behaviors, you just have the same behavior (or similar behavior because asynchronous seems to be handling retries differently for some reason). If you want to you can however have an interface with both synchronous and asynchronous operations which would allow to have different implementations (behaviors) in the future. \$\endgroup\$ – Bruno Costa Jan 16 '18 at 21:06
  • \$\begingroup\$ @david Yes, you would have to change all calls to doSendAsync passing null or a valid socket object \$\endgroup\$ – Bruno Costa Jan 16 '18 at 21:07
  • \$\begingroup\$ Yeah for sync we don't retry at all but for async we need to retry at all cost if acknowledgment is not received. Right now my sync is relying on async so that's why I thought maybe having a base class will work with common code and then having different implementations of it. \$\endgroup\$ – david Jan 16 '18 at 21:17

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.