<commit_before>const address = require('network-address'),
  url = require('url'),
  route = require('koa-route'),
  PromiseRepeat = require('promise-repeat');

import { mergeAttributes, createAttributes } from 'model-attributes';
import { ReceiveEndpoint } from 'kronos-endpoint';
import { Service, defineServiceConsumerProperties } from 'kronos-service';

function createWatchEndpoint(name, owner, makeWatch, dataProvider) {
  let watch;

  const options = {
    createOpposite: true
  };

  options.willBeClosed = () => {
    owner.trace({
      endpoint: this.identifier,
      state: 'close'
    });

    if (watch) {
      watch.end();
      watch = undefined;
    }
  };

  options.hasBeenOpened = () => {
    owner.trace({
      endpoint: this.identifier,
      state: 'open'
    });

    watch = makeWatch();
    watch.on('change', (data, res) => this.opposite.receive(data));
    watch.on('error', err =>
      owner.error({
        error: err,
        endpoint: this.identifier
      })
    );
  };

  const ep = new ReceiveEndpoint(name, owner, options);

  ep.receive = request => {
    if (request) {
      if (request.update && watch === undefined) {
      } else if (request.update === false && watch) {
        watch.end();
        watch = undefined;
      }
    }

    return dataProvider();
  };

  return ep;
}

export class ServiceConsul extends Service {
  static get name() {
    return 'consul';
  }

  get autostart() {
    return true;
  }

  constructor(config, owner) {
    super(config, owner);

    const svc = this;

    let watch;

    const nodesEndpoint = new ReceiveEndpoint('nodes', this, {
      createOpposite: true,
      willBeClosed() {
        this.trace({
          endpoint: this.identifier,
          state: 'close'
        });

        if (watch) {
          watch.end();
          watch = undefined;
        }
      },
      hasBeenOpened() {
        this.trace({
          endpoint: this.identifier,
          state: 'open'
        });
      }
    });

    nodesEndpoint.receive = request => {
      if (request) {
        if (request.update && watch === undefined) {
          watch = this.consul.watch({
            method: this.consul.catalog.service.nodes,
            options: {
              service: this.serviceDefinition.name
            }
          });

          watch.on('change', (data, res) => {
            console.log(data);
            return nodesEndpoint.opposite.receive(data);
          });
          watch.on('error', err => svc.error(err));
        } else if (request.update === false && watch) {
          watch.end();
          watch = undefined;
        }
      }

      return this.kronosNodes();
    };

    this.addEndpoint(nodesEndpoint);

    const options = {
      key: '',
      recurse: true
    };

    this.addEndpoint(
      createWatchEndpoint(
        'kv',
        this,
        () =>
          this.consul.watch({
            method: this.consul.kv.get,
            options: options
          }),
        async () => {
          const r = await this.consul.kv.get(options);
          console.log(`*** ***`);
          console.log(r);
          const dict = {};
          r[0].forEach(e => (dict[e.Key] = e.Value));
          //console.log(dict);
          return dict;
        }
      )
    );

    this.addEndpoint(
      createWatchEndpoint(
        'checks',
        this,
        () =>
          this.consul.watch({
            method: this.consul.health.checks,
            options: options
          }),
        () =>
          this.consul.health.checks(options).then(r => {
            console.log(r);
            return r;
          })
      )
    );
  }<commit_msg>fix: handle undefined in kv endpoint<commit_after>const address = require('network-address'),
  url = require('url'),
  route = require('koa-route'),
  PromiseRepeat = require('promise-repeat');

import { mergeAttributes, createAttributes } from 'model-attributes';
import { ReceiveEndpoint } from 'kronos-endpoint';
import { Service, defineServiceConsumerProperties } from 'kronos-service';

function createWatchEndpoint(name, owner, makeWatch, dataProvider) {
  let watch;

  const options = {
    createOpposite: true
  };

  options.willBeClosed = () => {
    owner.trace({
      endpoint: this.identifier,
      state: 'close'
    });

    if (watch) {
      watch.end();
      watch = undefined;
    }
  };

  options.hasBeenOpened = () => {
    owner.trace({
      endpoint: this.identifier,
      state: 'open'
    });

    watch = makeWatch();
    watch.on('change', (data, res) => this.opposite.receive(data));
    watch.on('error', err =>
      owner.error({
        error: err,
        endpoint: this.identifier
      })
    );
  };

  const ep = new ReceiveEndpoint(name, owner, options);

  ep.receive = request => {
    if (request) {
      if (request.update && watch === undefined) {
      } else if (request.update === false && watch) {
        watch.end();
        watch = undefined;
      }
    }

    return dataProvider();
  };

  return ep;
}

export class ServiceConsul extends Service {
  static get name() {
    return 'consul';
  }

  get autostart() {
    return true;
  }

  constructor(config, owner) {
    super(config, owner);

    const svc = this;

    let watch;

    const nodesEndpoint = new ReceiveEndpoint('nodes', this, {
      createOpposite: true,
      willBeClosed() {
        this.trace({
          endpoint: this.identifier,
          state: 'close'
        });

        if (watch) {
          watch.end();
          watch = undefined;
        }
      },
      hasBeenOpened() {
        this.trace({
          endpoint: this.identifier,
          state: 'open'
        });
      }
    });

    nodesEndpoint.receive = request => {
      if (request) {
        if (request.update && watch === undefined) {
          watch = this.consul.watch({
            method: this.consul.catalog.service.nodes,
            options: {
              service: this.serviceDefinition.name
            }
          });

          watch.on('change', (data, res) => {
            console.log(data);
            return nodesEndpoint.opposite.receive(data);
          });
          watch.on('error', err => svc.error(err));
        } else if (request.update === false && watch) {
          watch.end();
          watch = undefined;
        }
      }

      return this.kronosNodes();
    };

    this.addEndpoint(nodesEndpoint);

    const options = {
      key: '',
      recurse: true
    };

    this.addEndpoint(
      createWatchEndpoint(
        'kv',
        this,
        () =>
          this.consul.watch({
            method: this.consul.kv.get,
            options: options
          }),
        async () => {
          const r = await this.consul.kv.get(options);
          console.log(`*** ***`);
          console.log(r);
          const dict = {};
          if (r[0] !== undefined) {
            r[0].forEach(e => (dict[e.Key] = e.Value));
          }

          //console.log(dict);
          return dict;
        }
      )
    );

    this.addEndpoint(
      createWatchEndpoint(
        'checks',
        this,
        () =>
          this.consul.watch({
            method: this.consul.health.checks,
            options: options
          }),
        () =>
          this.consul.health.checks(options).then(r => {
            console.log(r);
            return r;
          })
      )
    );
  }
{'fraction_non_alphanumeric': 0.14158335584977033, 'fraction_numerical': 0.00040529586598216696, 'mean_word_length': 1.7489788340141106, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 31, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '3910621', 'n_tokens_mistral': 2183, 'n_tokens_neox': 2114, 'n_words': 436}