<commit_before>package org.dna.mqtt.moquette.messaging.spi.impl.subscriptions;

import java.text.ParseException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.LinkedBlockingDeque;

import org.dna.mqtt.moquette.messaging.spi.IStorageService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Represents a tree of topics subscriptions.
 *
 * @author andrea
 */
public class SubscriptionsStore {
    
    public static interface IVisitor<T> {
        void visit(TreeNode node);
        
        T getResult();
    }
    
    private class DumpTreeVisitor implements IVisitor<String> {
        
        String s = "";

        public void visit(TreeNode node) {
            String subScriptionsStr = "";
            for (Subscription sub : node.m_subscriptions) {
                subScriptionsStr += sub.toString();
            }
            s += node.getToken() == null ? "" : node.getToken().toString();
            s += subScriptionsStr + "\n";
        }
        
        public String getResult() {
            return s;
        }
    }
    
    private class SubscriptionTreeCollector implements IVisitor<List<Subscription>> {
        
        private List<Subscription> m_allSubscriptions = new ArrayList<Subscription>();

        public void visit(TreeNode node) {
            m_allSubscriptions.addAll(node.subscriptions());
        }
        
        public List<Subscription> getResult() {
            return m_allSubscriptions;
        }
    }

    private TreeNode subscriptions = new TreeNode(null);
    private static final Logger LOG = LoggerFactory.getLogger(SubscriptionsStore.class);

    private IStorageService m_storageService;

    /**
     * Initialize basic store structures, like the FS storage to maintain
     * client's topics subscriptions
     */
    public void init(IStorageService storageService) {
        LOG.debug("init invoked");

        m_storageService = storageService;

        //reload any subscriptions persisted
        if (LOG.isDebugEnabled()) {
            LOG.debug("subscription tree before {}", dumpTree());
        }
        LOG.debug("Reloading all stored subscriptions...");
        for (Subscription subscription : m_storageService.retrieveAllSubscriptions()) {
            LOG.debug("Re-subscribing {} to topic {}", subscription.getClientId(), subscription.getTopic());
            addDirect(subscription);
        }
        if (LOG.isDebugEnabled()) {
            LOG.debug("subscription tree after {}", dumpTree());
        }
        LOG.debug("Finished loading");
    }
    
    protected void addDirect(Subscription newSubscription) {<commit_msg>Minor change, condensed debug logging<commit_after>package org.dna.mqtt.moquette.messaging.spi.impl.subscriptions;

import java.text.ParseException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.LinkedBlockingDeque;

import org.dna.mqtt.moquette.messaging.spi.IStorageService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Represents a tree of topics subscriptions.
 *
 * @author andrea
 */
public class SubscriptionsStore {
    
    public static interface IVisitor<T> {
        void visit(TreeNode node);
        
        T getResult();
    }
    
    private class DumpTreeVisitor implements IVisitor<String> {
        
        String s = "";

        public void visit(TreeNode node) {
            String subScriptionsStr = "";
            for (Subscription sub : node.m_subscriptions) {
                subScriptionsStr += sub.toString();
            }
            s += node.getToken() == null ? "" : node.getToken().toString();
            s += subScriptionsStr + "\n";
        }
        
        public String getResult() {
            return s;
        }
    }
    
    private class SubscriptionTreeCollector implements IVisitor<List<Subscription>> {
        
        private List<Subscription> m_allSubscriptions = new ArrayList<Subscription>();

        public void visit(TreeNode node) {
            m_allSubscriptions.addAll(node.subscriptions());
        }
        
        public List<Subscription> getResult() {
            return m_allSubscriptions;
        }
    }

    private TreeNode subscriptions = new TreeNode(null);
    private static final Logger LOG = LoggerFactory.getLogger(SubscriptionsStore.class);

    private IStorageService m_storageService;

    /**
     * Initialize basic store structures, like the FS storage to maintain
     * client's topics subscriptions
     */
    public void init(IStorageService storageService) {
        LOG.debug("init invoked");

        m_storageService = storageService;

        //reload any subscriptions persisted
        if (LOG.isDebugEnabled()) {
            LOG.debug("Reloading all stored subscriptions...subscription tree before {}", dumpTree());
        }
        
        for (Subscription subscription : m_storageService.retrieveAllSubscriptions()) {
            LOG.debug("Re-subscribing {} to topic {}", subscription.getClientId(), subscription.getTopic());
            addDirect(subscription);
        }
        if (LOG.isDebugEnabled()) {
            LOG.debug("Finished loading. Subscription tree after {}", dumpTree());
        }
    }
    
    protected void addDirect(Subscription newSubscription) {
{'fraction_non_alphanumeric': 0.09894385769872151, 'fraction_numerical': 0.0007411524921252547, 'mean_word_length': 2.5143229166666665, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 17, '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': '2659276', 'n_tokens_mistral': 1408, 'n_tokens_neox': 1407, 'n_words': 327}