<commit_before>/**
 * @license
 * Copyright 2019 The FOAM Authors. All Rights Reserved.
 * http://www.apache.org/licenses/LICENSE-2.0
 */

foam.CLASS({
  package: 'foam.nanos.medusa',
  name: 'ClusterConfigPingSink',
  extends: 'foam.dao.AbstractSink',

  javaImports: [
    'foam.dao.DAO',
    'static foam.mlang.MLang.EQ',
    'static foam.mlang.MLang.MAX',
    'foam.mlang.sink.Max',
    'foam.nanos.http.PingService',
    'foam.nanos.logger.PrefixLogger',
    'foam.nanos.logger.Logger'
  ],

  axioms: [
    {
      buildJavaClass: function(cls) {
        cls.extras.push(`
          public ClusterConfigPingSink(foam.core.X x, foam.dao.DAO dao, int timeout) {
            setX(x);
            setDao(dao);
            setTimeout(timeout);
          }
        `);
      }
    }
  ],

  properties: [
    {
      name: 'timeout',
      class: 'Int',
      value: 3000,
    },
    {
      name: 'dao',
      class: 'foam.dao.DAOProperty'
    },
    {
      name: 'logger',
      class: 'FObjectProperty',
      of: 'foam.nanos.logger.Logger',
      visibility: 'HIDDEN',
      javaFactory: `
        return new PrefixLogger(new Object[] {
          this.getClass().getSimpleName()
        }, (Logger) getX().get("logger"));
      `
    }
 ],

  methods: [
    {
      name: 'put',
      args: [
        {
          name: 'obj',
          type: 'Object'
        },
        {
          name: 'sub',
          type: 'foam.core.Detachable'
        }
      ],
      javaCode: `
      DaggerService dagger = (DaggerService) getX().get("daggerService");
      ClusterConfigService service = (ClusterConfigService) getX().get("clusterConfigService");
      ClusterConfig myConfig = service.getConfig(getX(), service.getConfigId());
      ClusterConfig config = (ClusterConfig) ((ClusterConfig) obj).fclone();
      PingService ping = (PingService) getX().get("ping");

      try {
        Long latency = ping.ping(getX(), config.getId(), config.getPort(), getTimeout());
        config.setPingLatency(latency);
        if ( config.getStatus() != Status.ONLINE) {
          getLogger().info(config.getName(), config.getType().getLabel(), config.getStatus().getLabel(), "->", "ONLINE");
          config.setStatus(Status.ONLINE);
          config = (ClusterConfig) getDao().put_(getX(), config);

          // If a Node comming online, begin replay from it.
          if ( myConfig.getType() == MedusaType.MEDIATOR &&
               config.getType() == MedusaType.NODE &&
               config.getZone() == 0L &&
               config.getRegion() == myConfig.getRegion() &&
               config.getRealm() == myConfig.getRealm() ) {

            DAO clientDAO = service.getClientDAO(getX(), "medusaEntryDAO", myConfig, config);
            clientDAO = new RetryClientSinkDAO.Builder(getX())
              .setDelegate(clientDAO)
              .setMaxRetryAttempts(service.getMaxRetryAttempts())
              .setMaxRetryDelay(service.getMaxRetryDelay())
              .build();

            ReplayDetailsCmd details = new ReplayDetailsCmd();
            details.setRequester(myConfig.getId());
            details.setResponder(config.getId());
            getLogger().debug(myConfig.getId(), "ReplayDetailsCmd to", config.getId());
            details = (ReplayDetailsCmd) clientDAO.cmd_(getX(), details);
            getLogger().debug(myConfig.getId(), "ReplayDetailsCmd from", config.getId(), details);

            dagger.setGlobalIndex(getX(), details.getMaxIndex());

            // Send to Consensus DAO to prepare for Replay
            ((DAO) getX().get("medusaEntryDAO")).cmd(details);

            // NOTE: using internalMedusaEntryDAO else we'll block on ReplayingDAO.
            DAO dao = (DAO) getX().get("internalMedusaEntryDAO");
            dao = dao.where(EQ(MedusaEntry.HAS_CONSENSUS, true));<commit_msg>clear ping info on ONLINE<commit_after>/**
 * @license
 * Copyright 2019 The FOAM Authors. All Rights Reserved.
 * http://www.apache.org/licenses/LICENSE-2.0
 */

foam.CLASS({
  package: 'foam.nanos.medusa',
  name: 'ClusterConfigPingSink',
  extends: 'foam.dao.AbstractSink',

  javaImports: [
    'foam.dao.DAO',
    'static foam.mlang.MLang.EQ',
    'static foam.mlang.MLang.MAX',
    'foam.mlang.sink.Max',
    'foam.nanos.http.PingService',
    'foam.nanos.logger.PrefixLogger',
    'foam.nanos.logger.Logger'
  ],

  axioms: [
    {
      buildJavaClass: function(cls) {
        cls.extras.push(`
          public ClusterConfigPingSink(foam.core.X x, foam.dao.DAO dao, int timeout) {
            setX(x);
            setDao(dao);
            setTimeout(timeout);
          }
        `);
      }
    }
  ],

  properties: [
    {
      name: 'timeout',
      class: 'Int',
      value: 3000,
    },
    {
      name: 'dao',
      class: 'foam.dao.DAOProperty'
    },
    {
      name: 'logger',
      class: 'FObjectProperty',
      of: 'foam.nanos.logger.Logger',
      visibility: 'HIDDEN',
      javaFactory: `
        return new PrefixLogger(new Object[] {
          this.getClass().getSimpleName()
        }, (Logger) getX().get("logger"));
      `
    }
 ],

  methods: [
    {
      name: 'put',
      args: [
        {
          name: 'obj',
          type: 'Object'
        },
        {
          name: 'sub',
          type: 'foam.core.Detachable'
        }
      ],
      javaCode: `
      DaggerService dagger = (DaggerService) getX().get("daggerService");
      ClusterConfigService service = (ClusterConfigService) getX().get("clusterConfigService");
      ClusterConfig myConfig = service.getConfig(getX(), service.getConfigId());
      ClusterConfig config = (ClusterConfig) ((ClusterConfig) obj).fclone();
      PingService ping = (PingService) getX().get("ping");

      try {
        Long latency = ping.ping(getX(), config.getId(), config.getPort(), getTimeout());
        config.setPingLatency(latency);
        if ( config.getStatus() != Status.ONLINE) {
          getLogger().info(config.getName(), config.getType().getLabel(), config.getStatus().getLabel(), "->", "ONLINE");
          config.setStatus(Status.ONLINE);
          config.setPingInfo("");
          config = (ClusterConfig) getDao().put_(getX(), config);

          // If a Node comming online, begin replay from it.
          if ( myConfig.getType() == MedusaType.MEDIATOR &&
               config.getType() == MedusaType.NODE &&
               config.getZone() == 0L &&
               config.getRegion() == myConfig.getRegion() &&
               config.getRealm() == myConfig.getRealm() ) {

            DAO clientDAO = service.getClientDAO(getX(), "medusaEntryDAO", myConfig, config);
            clientDAO = new RetryClientSinkDAO.Builder(getX())
              .setDelegate(clientDAO)
              .setMaxRetryAttempts(service.getMaxRetryAttempts())
              .setMaxRetryDelay(service.getMaxRetryDelay())
              .build();

            ReplayDetailsCmd details = new ReplayDetailsCmd();
            details.setRequester(myConfig.getId());
            details.setResponder(config.getId());
            getLogger().debug(myConfig.getId(), "ReplayDetailsCmd to", config.getId());
            details = (ReplayDetailsCmd) clientDAO.cmd_(getX(), details);
            getLogger().debug(myConfig.getId(), "ReplayDetailsCmd from", config.getId(), details);

            dagger.setGlobalIndex(getX(), details.getMaxIndex());

            // Send to Consensus DAO to prepare for Replay
            ((DAO) getX().get("medusaEntryDAO")).cmd(details);

            // NOTE: using internalMedusaEntryDAO else we'll block on ReplayingDAO.
            DAO dao = (DAO) getX().get("internalMedusaEntryDAO");
            dao = dao.where(EQ(MedusaEntry.HAS_CONSENSUS, true));
{'fraction_non_alphanumeric': 0.14813850559750066, 'fraction_numerical': 0.002863837542306691, 'mean_word_length': 2.3073611708996986, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6807456', 'n_tokens_mistral': 2394, 'n_tokens_neox': 2387, 'n_words': 408}