<commit_before>library wamp.server;

import 'dart:io';
import 'dart:json' as JSON;
import 'dart:math';
import 'dart:async';
import 'package:wamp/wamp.dart';

part 'src/server/client.dart';

class WampHandler {
  Set<Client> clients = new Set();
  Map<String, Set<Client>> topicMap = new Map();
  CurieCodec curie = new CurieCodec();

  onOpen(WebSocketConnection conn) {
    var c = new Client(conn)
        ..welcome();

    clients.add(c);

    conn.onClosed = (int status, String reason) {
      clients.remove(c);
    };

    conn.onMessage = (m) {
      var msg = JSON.parse(m);

      switch(msg[0]) {
        case MessageType.PREFIX:
          c.prefixes[msg[1]] = msg[2];
          break;

        case MessageType.CALL:
          onCall(c, msg[1], msg[2], msg[3]);
          break;

        case MessageType.SUBSCRIBE:
          onSubscribe(c, msg[1]);
          break;

        case MessageType.UNSUBSCRIBE:
          onUnsubscribe(c, msg[1]);
          break;

        case MessageType.PUBLISH:
          onPublish(c, msg[1], msg[2]/*, msg[3], msg[4]*/);
          break;
      }

    };

//    conn.onClosed = (e) {
//      clients.remove(c);
//    };
  }

  void onCall(Client c, String callId, String uri, arg) {
  }

  void onSubscribe(Client c, String topicUri) {
    var uri = curie.decode(topicUri);

    c.topics.add(uri);

    if (!topicMap.containsKey(uri)) {
      topicMap[uri] = new Set();
    }
    topicMap[uri].add(c);
  }

  void onUnsubscribe(Client c, String topicUri) {
    var uri = curie.decode(topicUri);

    c.topics.remove(uri);

    if (topicMap.containsKey(uri)) {
      topicMap[uri].remove(c);
      if (topicMap[uri].isEmpty) topicMap.remove(uri);
    }
  }

  void onPublish(Client c, String topicUri, event, [exclude, eligible]) {
//    var uri = curie.decode(topicUri);
//
//    clients.forEach((client) {
//      client.event(topicUri, event);
//    });

    publish(topicUri, event);
  }

  /**
   * Publish an event to all the subscribed clients.
   */
  void publish(String topicUri, event) {
    final uri = curie.decode(topicUri),
          subscribers = topicMap[uri];

    subscribers.forEach((client) {
      client.event(topicUri, event);
    });
  }
}<commit_msg>Unsubscribe disconnected clients.<commit_after>library wamp.server;

import 'dart:io';
import 'dart:json' as JSON;
import 'dart:math';
import 'dart:async';
import 'package:wamp/wamp.dart';

part 'src/server/client.dart';

/**
 * Handler for wamp connections.
 */
class WampHandler {
  Set<Client> clients = new Set();
  Map<String, Set<Client>> topicMap = new Map();
  CurieCodec curie = new CurieCodec();

  onOpen(WebSocketConnection conn) {
    var c = new Client(conn)
        ..welcome();

    clients.add(c);

    conn.onClosed = (int status, String reason) {
      clients.remove(c);
    };

    conn.onMessage = (m) {
      var msg = JSON.parse(m);

      switch(msg[0]) {
        case MessageType.PREFIX:
          c.prefixes[msg[1]] = msg[2];
          break;

        case MessageType.CALL:
          onCall(c, msg[1], msg[2], msg[3]);
          break;

        case MessageType.SUBSCRIBE:
          onSubscribe(c, msg[1]);
          break;

        case MessageType.UNSUBSCRIBE:
          onUnsubscribe(c, msg[1]);
          break;

        case MessageType.PUBLISH:
          onPublish(c, msg[1], msg[2]/*, msg[3], msg[4]*/);
          break;
      }

    };
  }

  void onCall(Client c, String callId, String uri, arg) {
  }

  void onSubscribe(Client c, String topicUri) {
    var uri = curie.decode(topicUri);

    c.topics.add(uri);

    if (!topicMap.containsKey(uri)) {
      topicMap[uri] = new Set();
    }
    topicMap[uri].add(c);
  }

  void onUnsubscribe(Client c, String topicUri) {
    var uri = curie.decode(topicUri);

    c.topics.remove(uri);

    if (topicMap.containsKey(uri)) {
      topicMap[uri].remove(c);
      if (topicMap[uri].isEmpty) topicMap.remove(uri);
    }
  }

  void onPublish(Client c, String topicUri, event, [exclude, eligible]) {
    // TODO: handle exclude, eligible.
    publish(topicUri, event);
  }

  /**
   * Publish an event to all the subscribed clients.
   */
  void publish(String topicUri, event) {
    final uri = curie.decode(topicUri),
          subscribers = topicMap[uri],
          garbage = [];

    subscribers.forEach((client) {
      if (clients.contains(client)) {
        client.event(topicUri, event);
      } else {
        // The client is disconnected, so gc.
        garbage.add(client);
      }
    });

    // TODO: perfome gc when the client is disconnected?
    if (!garbage.isEmpty) {
      garbage.forEach((c) => clients.remove(c));
    }
  }
}
{'fraction_non_alphanumeric': 0.14795698924731182, 'fraction_numerical': 0.005161290322580645, 'mean_word_length': 2.44008875739645, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 10, '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': '2444554', 'n_tokens_mistral': 1617, 'n_tokens_neox': 1574, 'n_words': 340}