publisherApp.java

import java.util.Random;
import org.zeromq.ZMQ;

class publisherApp
{
  public static void main(String[] args) throws InterruptedException
  {
    // Initialize 0MQ with a single I/O thread,
    ZMQ.Context ctx = ZMQ.context(1);

    // Create a PUB socket for port 5555 on the loopback interface
    ZMQ.Socket s = ctx.socket(ZMQ.PUB);
    s.connect("tcp://127.0.0.1:5555");
    s.setHWM(1); // High water mark for when to throw away messages
    s.setSwap(25000000); // Swap space on disk in bytes

    Random rand = new Random();

    for (long msg_id = 1; msg_id < Long.MAX_VALUE; msg_id++)
    {
      // Create a new, empty, 8 byte message
      int zipcode, temperature, relhumidity;
      zipcode = rand.nextInt(100000);
      temperature = rand.nextInt(215) - 80;
      relhumidity = rand.nextInt(50) + 10;

      String zip = String.format("%05d", zipcode);
      Zmqtest.Weather weather = Zmqtest.Weather.newBuilder().setZipcode(zip).setTemperature(temperature)
          .setRelhumidity(relhumidity).build();

      // System.out.println("sending: " + info);

      // Publish our message
      s.send(zip.getBytes(), ZMQ.SNDMORE);
      s.send(weather.toByteArray(), 0);
    }
  }
}

subscriberApp.java

import org.zeromq.ZMQ;
import com.google.protobuf.InvalidProtocolBufferException;

public class subscriberApp
{
  public static void main(String[] args) throws InvalidProtocolBufferException
  {
    // Initialize 0MQ with a single I/O thread,
    ZMQ.Context ctx = ZMQ.context(1);

    // Create a PUB socket for port 5555 on the loopback interface
    ZMQ.Socket s = ctx.socket(ZMQ.SUB);
    byte[] filter = "21600".getBytes();
    s.subscribe(filter);

    s.bind("tcp://127.0.0.1:5555");

    long update_nbr;
    long total_temp = 0;
    for (update_nbr = 0; update_nbr < 15; update_nbr++)
    {
      s.recv(0); // Ignore the topic
      final byte[] recv = s.recv(0);
      final Zmqtest.Weather weather = Zmqtest.Weather.parseFrom(recv);
      total_temp += weather.getTemperature();

//      if (update_nbr % 1000 == 0)
//      {
        System.out.println(update_nbr + " " + total_temp);
//      }
    }

    System.out.format(
        "Average temperature for filter '%s' was %dF\n",
        new String(filter),
        (int) (total_temp / update_nbr));
  }
}

Zmqtest.java

// Generated by the protocol buffer compiler.  DO NOT EDIT!
// source: gen/zmqtest.proto



public final class Zmqtest {
  private Zmqtest() {}
  public static void registerAllExtensions(
      com.google.protobuf.ExtensionRegistry registry) {
  }
  public static final class Weather extends
      com.google.protobuf.GeneratedMessage {
    // Use Weather.newBuilder() to construct.
    private Weather() {
      initFields();
    }
    private Weather(boolean noInit) {}
    
    private static final Weather defaultInstance;
    public static Weather getDefaultInstance() {
      return defaultInstance;
    }
    
    public Weather getDefaultInstanceForType() {
      return defaultInstance;
    }
    
    public static final com.google.protobuf.Descriptors.Descriptor
        getDescriptor() {
      return Zmqtest.internal_static_com_lmco_sts_srl_zmqtest_Weather_descriptor;
    }
    
    protected com.google.protobuf.GeneratedMessage.FieldAccessorTable
        internalGetFieldAccessorTable() {
      return Zmqtest.internal_static_com_lmco_sts_srl_zmqtest_Weather_fieldAccessorTable;
    }
    
    // required string zipcode = 1;
    public static final int ZIPCODE_FIELD_NUMBER = 1;
    private boolean hasZipcode;
    private java.lang.String zipcode_ = "";
    public boolean hasZipcode() { return hasZipcode; }
    public java.lang.String getZipcode() { return zipcode_; }
    
    // required int32 temperature = 2;
    public static final int TEMPERATURE_FIELD_NUMBER = 2;
    private boolean hasTemperature;
    private int temperature_ = 0;
    public boolean hasTemperature() { return hasTemperature; }
    public int getTemperature() { return temperature_; }
    
    // optional int32 relhumidity = 3;
    public static final int RELHUMIDITY_FIELD_NUMBER = 3;
    private boolean hasRelhumidity;
    private int relhumidity_ = 0;
    public boolean hasRelhumidity() { return hasRelhumidity; }
    public int getRelhumidity() { return relhumidity_; }
    
    private void initFields() {
    }
    public final boolean isInitialized() {
      if (!hasZipcode) return false;
      if (!hasTemperature) return false;
      return true;
    }
    
    public void writeTo(com.google.protobuf.CodedOutputStream output)
                        throws java.io.IOException {
      getSerializedSize();
      if (hasZipcode()) {
        output.writeString(1, getZipcode());
      }
      if (hasTemperature()) {
        output.writeInt32(2, getTemperature());
      }
      if (hasRelhumidity()) {
        output.writeInt32(3, getRelhumidity());
      }
      getUnknownFields().writeTo(output);
    }
    
    private int memoizedSerializedSize = -1;
    public int getSerializedSize() {
      int size = memoizedSerializedSize;
      if (size != -1) return size;
    
      size = 0;
      if (hasZipcode()) {
        size += com.google.protobuf.CodedOutputStream
          .computeStringSize(1, getZipcode());
      }
      if (hasTemperature()) {
        size += com.google.protobuf.CodedOutputStream
          .computeInt32Size(2, getTemperature());
      }
      if (hasRelhumidity()) {
        size += com.google.protobuf.CodedOutputStream
          .computeInt32Size(3, getRelhumidity());
      }
      size += getUnknownFields().getSerializedSize();
      memoizedSerializedSize = size;
      return size;
    }
    
    public static Zmqtest.Weather parseFrom(
        com.google.protobuf.ByteString data)
        throws com.google.protobuf.InvalidProtocolBufferException {
      return newBuilder().mergeFrom(data).buildParsed();
    }
    public static Zmqtest.Weather parseFrom(
        com.google.protobuf.ByteString data,
        com.google.protobuf.ExtensionRegistryLite extensionRegistry)
        throws com.google.protobuf.InvalidProtocolBufferException {
      return newBuilder().mergeFrom(data, extensionRegistry)
               .buildParsed();
    }
    public static Zmqtest.Weather parseFrom(byte[] data)
        throws com.google.protobuf.InvalidProtocolBufferException {
      return newBuilder().mergeFrom(data).buildParsed();
    }
    public static Zmqtest.Weather parseFrom(
        byte[] data,
        com.google.protobuf.ExtensionRegistryLite extensionRegistry)
        throws com.google.protobuf.InvalidProtocolBufferException {
      return newBuilder().mergeFrom(data, extensionRegistry)
               .buildParsed();
    }
    public static Zmqtest.Weather parseFrom(java.io.InputStream input)
        throws java.io.IOException {
      return newBuilder().mergeFrom(input).buildParsed();
    }
    public static Zmqtest.Weather parseFrom(
        java.io.InputStream input,
        com.google.protobuf.ExtensionRegistryLite extensionRegistry)
        throws java.io.IOException {
      return newBuilder().mergeFrom(input, extensionRegistry)
               .buildParsed();
    }
    public static Zmqtest.Weather parseDelimitedFrom(java.io.InputStream input)
        throws java.io.IOException {
      Builder builder = newBuilder();
      if (builder.mergeDelimitedFrom(input)) {
        return builder.buildParsed();
      } else {
        return null;
      }
    }
    public static Zmqtest.Weather parseDelimitedFrom(
        java.io.InputStream input,
        com.google.protobuf.ExtensionRegistryLite extensionRegistry)
        throws java.io.IOException {
      Builder builder = newBuilder();
      if (builder.mergeDelimitedFrom(input, extensionRegistry)) {
        return builder.buildParsed();
      } else {
        return null;
      }
    }
    public static Zmqtest.Weather parseFrom(
        com.google.protobuf.CodedInputStream input)
        throws java.io.IOException {
      return newBuilder().mergeFrom(input).buildParsed();
    }
    public static Zmqtest.Weather parseFrom(
        com.google.protobuf.CodedInputStream input,
        com.google.protobuf.ExtensionRegistryLite extensionRegistry)
        throws java.io.IOException {
      return newBuilder().mergeFrom(input, extensionRegistry)
               .buildParsed();
    }
    
    public static Builder newBuilder() { return Builder.create(); }
    public Builder newBuilderForType() { return newBuilder(); }
    public static Builder newBuilder(Zmqtest.Weather prototype) {
      return newBuilder().mergeFrom(prototype);
    }
    public Builder toBuilder() { return newBuilder(this); }
    
    public static final class Builder extends
        com.google.protobuf.GeneratedMessage.Builder<Builder> {
      private Zmqtest.Weather result;
      
      // Construct using com.lmco.sts.srl.zmqtest.Zmqtest.Weather.newBuilder()
      private Builder() {}
      
      private static Builder create() {
        Builder builder = new Builder();
        builder.result = new Zmqtest.Weather();
        return builder;
      }
      
      protected Zmqtest.Weather internalGetResult() {
        return result;
      }
      
      public Builder clear() {
        if (result == null) {
          throw new IllegalStateException(
            "Cannot call clear() after build().");
        }
        result = new Zmqtest.Weather();
        return this;
      }
      
      public Builder clone() {
        return create().mergeFrom(result);
      }
      
      public com.google.protobuf.Descriptors.Descriptor
          getDescriptorForType() {
        return Zmqtest.Weather.getDescriptor();
      }
      
      public Zmqtest.Weather getDefaultInstanceForType() {
        return Zmqtest.Weather.getDefaultInstance();
      }
      
      public boolean isInitialized() {
        return result.isInitialized();
      }
      public Zmqtest.Weather build() {
        if (result != null && !isInitialized()) {
          throw newUninitializedMessageException(result);
        }
        return buildPartial();
      }
      
      private Zmqtest.Weather buildParsed()
          throws com.google.protobuf.InvalidProtocolBufferException {
        if (!isInitialized()) {
          throw newUninitializedMessageException(
            result).asInvalidProtocolBufferException();
        }
        return buildPartial();
      }
      
      public Zmqtest.Weather buildPartial() {
        if (result == null) {
          throw new IllegalStateException(
            "build() has already been called on this Builder.");
        }
        Zmqtest.Weather returnMe = result;
        result = null;
        return returnMe;
      }
      
      public Builder mergeFrom(com.google.protobuf.Message other) {
        if (other instanceof Zmqtest.Weather) {
          return mergeFrom((Zmqtest.Weather)other);
        } else {
          super.mergeFrom(other);
          return this;
        }
      }
      
      public Builder mergeFrom(Zmqtest.Weather other) {
        if (other == Zmqtest.Weather.getDefaultInstance()) return this;
        if (other.hasZipcode()) {
          setZipcode(other.getZipcode());
        }
        if (other.hasTemperature()) {
          setTemperature(other.getTemperature());
        }
        if (other.hasRelhumidity()) {
          setRelhumidity(other.getRelhumidity());
        }
        this.mergeUnknownFields(other.getUnknownFields());
        return this;
      }
      
      public Builder mergeFrom(
          com.google.protobuf.CodedInputStream input,
          com.google.protobuf.ExtensionRegistryLite extensionRegistry)
          throws java.io.IOException {
        com.google.protobuf.UnknownFieldSet.Builder unknownFields =
          com.google.protobuf.UnknownFieldSet.newBuilder(
            this.getUnknownFields());
        while (true) {
          int tag = input.readTag();
          switch (tag) {
            case 0:
              this.setUnknownFields(unknownFields.build());
              return this;
            default: {
              if (!parseUnknownField(input, unknownFields,
                                     extensionRegistry, tag)) {
                this.setUnknownFields(unknownFields.build());
                return this;
              }
              break;
            }
            case 10: {
              setZipcode(input.readString());
              break;
            }
            case 16: {
              setTemperature(input.readInt32());
              break;
            }
            case 24: {
              setRelhumidity(input.readInt32());
              break;
            }
          }
        }
      }
      
      
      // required string zipcode = 1;
      public boolean hasZipcode() {
        return result.hasZipcode();
      }
      public java.lang.String getZipcode() {
        return result.getZipcode();
      }
      public Builder setZipcode(java.lang.String value) {
        if (value == null) {
    throw new NullPointerException();
  }
  result.hasZipcode = true;
        result.zipcode_ = value;
        return this;
      }
      public Builder clearZipcode() {
        result.hasZipcode = false;
        result.zipcode_ = getDefaultInstance().getZipcode();
        return this;
      }
      
      // required int32 temperature = 2;
      public boolean hasTemperature() {
        return result.hasTemperature();
      }
      public int getTemperature() {
        return result.getTemperature();
      }
      public Builder setTemperature(int value) {
        result.hasTemperature = true;
        result.temperature_ = value;
        return this;
      }
      public Builder clearTemperature() {
        result.hasTemperature = false;
        result.temperature_ = 0;
        return this;
      }
      
      // optional int32 relhumidity = 3;
      public boolean hasRelhumidity() {
        return result.hasRelhumidity();
      }
      public int getRelhumidity() {
        return result.getRelhumidity();
      }
      public Builder setRelhumidity(int value) {
        result.hasRelhumidity = true;
        result.relhumidity_ = value;
        return this;
      }
      public Builder clearRelhumidity() {
        result.hasRelhumidity = false;
        result.relhumidity_ = 0;
        return this;
      }
      
      // @@protoc_insertion_point(builder_scope:com.lmco.sts.srl.zmqtest.Weather)
    }
    
    static {
      defaultInstance = new Weather(true);
      Zmqtest.internalForceInit();
      defaultInstance.initFields();
    }
    
    // @@protoc_insertion_point(class_scope:com.lmco.sts.srl.zmqtest.Weather)
  }
  
  private static com.google.protobuf.Descriptors.Descriptor
    internal_static_com_lmco_sts_srl_zmqtest_Weather_descriptor;
  private static
    com.google.protobuf.GeneratedMessage.FieldAccessorTable
      internal_static_com_lmco_sts_srl_zmqtest_Weather_fieldAccessorTable;
  
  public static com.google.protobuf.Descriptors.FileDescriptor
      getDescriptor() {
    return descriptor;
  }
  private static com.google.protobuf.Descriptors.FileDescriptor
      descriptor;
  static {
    java.lang.String[] descriptorData = {
      "\n\021gen/zmqtest.proto\022\030com.lmco.sts.srl.zm" +
      "qtest\"D\n\007Weather\022\017\n\007zipcode\030\001 \002(\t\022\023\n\013tem" +
      "perature\030\002 \002(\005\022\023\n\013relhumidity\030\003 \001(\005"
    };
    com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner assigner =
      new com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner() {
        public com.google.protobuf.ExtensionRegistry assignDescriptors(
            com.google.protobuf.Descriptors.FileDescriptor root) {
          descriptor = root;
          internal_static_com_lmco_sts_srl_zmqtest_Weather_descriptor =
            getDescriptor().getMessageTypes().get(0);
          internal_static_com_lmco_sts_srl_zmqtest_Weather_fieldAccessorTable = new
            com.google.protobuf.GeneratedMessage.FieldAccessorTable(
              internal_static_com_lmco_sts_srl_zmqtest_Weather_descriptor,
              new java.lang.String[] { "Zipcode", "Temperature", "Relhumidity", },
              Zmqtest.Weather.class,
              Zmqtest.Weather.Builder.class);
          return null;
        }
      };
    com.google.protobuf.Descriptors.FileDescriptor
      .internalBuildGeneratedFileFrom(descriptorData,
        new com.google.protobuf.Descriptors.FileDescriptor[] {
        }, assigner);
  }
  
  public static void internalForceInit() {}
  
  // @@protoc_insertion_point(outer_class_scope)
}