brachi-wernick

public class FieldsConfigurationProvider{

    private Logger logger = LoggerFactory.getLogger(FieldsConfigurationProvider.class);
    private Long lastUpdateTime;
    private FieldsConfiguration config;

    private String bucketName;
    private  String filePath;
    private  long fieldsConfigLoadInterval;

brachi-wernick

private void watch() {
        new Thread(() -> {

            Storage storage = StorageOptions.getDefaultInstance().getService();
            Bucket bucket = storage.get(bucketName);

            while(true) {
                try {
                    Blob blob = bucket.get(filePath);
                    long currentTime = blob.getUpdateTime();

brachi-wernick

 @ProcessElement
  public void processElement(@Element JsonNode jsonNode, ProcessContext context, PipelineOptions options) {

        FieldConfigurationOptions fieldConfigurationOptions = options.as(FieldConfigurationOptions.class);

        FieldConfigurationLoader fieldsConfigurationProvider = fieldConfigurationOptions.getFieldsConfigurationProvider();

        FieldsConfiguration config = fieldsConfigurationProvider.getConfig();
    
        TableRow tableRow = new TableRow();

brachi-wernick

public class MyFn extends DoFn<String, String> {

    private  static RedissonClient redisClient;
    private final static Object lock = new Object();
    private String redisHost;
    private int redisPort;

  
   MyFn(String redisHost, Integer redisPort) {
        this.redisHost = redisHost;

brachi-wernick

public interface RedisOptions extends PipelineOptions {

    ValueProvider<String> getRedisHost();

    void setRedisHost(ValueProvider<String> value);

    ValueProvider<Integer> getRedisPort();
    void setRedisPort(ValueProvider<Integer> value);

brachi-wernick

public class RedisClientFactory implements DefaultValueFactory<RedissonClient> {
    @Override
    public RedissonClient create(PipelineOptions options) {
        RedisOptions redisOptions = options.as(RedisOptions.class);
        
        Config config = new Config();

        config.useSingleServer()
                .setAddress("redis://" + redisOptions.getRedisHost().get() + ":" + redisOptions.getRedisPort().get()");

brachi-wernick

public class MyFn extends DoFn<String, String> {
    
  @ProcessElement
   public void processElement(ProcessContext context,PipelineOptions options) {
      
     RedisOptions redisOptions = options.as(RedisOptions.class);
     RedissonClient redissonClient = redisOptions.getRedissonClient();
    
    //some logic enrichment
    redis.get() 

brachi-wernick

 public class MainVerticle extends AbstractVerticle
{

 
   @Override
    public void start(Promise<Void> startPromise) throws Exception {

        Router router = Router.router(vertx);
        router.get("/hello").handler(this::hello);

brachi-wernick

FROM java:8-jre
ENV VERTICLE_FILE starter-1.0.0-SNAPSHOT-fat.jar
# Set the location of the verticles
ENV VERTICLE_HOME /usr/verticles
EXPOSE 8882
# Copy your fat jar to the container
COPY target/$VERTICLE_FILE $VERTICLE_HOME/
# Launch the verticle
WORKDIR $VERTICLE_HOME
ENTRYPOINT ["sh", "-c"]

brachi-wernick

public class RedisClient {


    private static RedissonClient redisClient;

    public static synchronized void newInstance(){
        if(redisClient==null) {
            Config config = new Config();
            config.setCodec(new StringCodec());
            config.setThreads(Integer.parseInt(System.getenv("threads")));