<commit_before>/*
 * #%L
 * Wisdom-Framework
 * %%
 * Copyright (C) 2013 - 2014 Wisdom Framework
 * %%
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * 
 *      http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 * #L%
 */
package org.wisdom.wamp;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ArrayNode;
import org.apache.felix.ipojo.annotations.*;
import org.osgi.service.event.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.wisdom.api.Controller;
import org.wisdom.api.DefaultController;
import org.wisdom.api.annotations.*;
import org.wisdom.api.content.Json;
import org.wisdom.api.engine.WisdomEngine;
import org.wisdom.api.http.websockets.Publisher;
import org.wisdom.wamp.messages.*;
import org.wisdom.wamp.messages.Event;
import org.wisdom.wamp.services.ExportedService;
import org.wisdom.wamp.services.RegistryException;
import org.wisdom.wamp.services.Wamp;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.*;

/**
 * The WAMP main controller.
 * This controller is responsible for receiving the WAMP messages and sending the responses.
 * Components willing to be exposed using the Wamp protocol must register themselves using the WampRegistry service.
 */
@Component(immediate = true)
@Provides(specifications = {Wamp.class, EventHandler.class, Controller.class})
@Instantiate
public class WampController extends DefaultController implements Wamp, EventHandler {

    private static final Logger LOGGER = LoggerFactory.getLogger(WampController.class);

    private final String prefix;
    private final String errorPrefix;

    @Requires
    Publisher publisher;

    @Requires
    Json json;

    @Requires(optional = true)
    EventAdmin ea;

    /**
     * The topics from the event admin listened by this component to transfer the event to WAMP.
     */
    @ServiceProperty(name=EventConstants.EVENT_TOPIC)
    String[] topics = new String[] {
            "wamp/*"
    };

    /**
     * To avoid loop, we ignore all event send by us. The event we sent have a 'wamp.topic' property,
     * to we accept only event without this property.
     */
    @ServiceProperty(name=EventConstants.EVENT_FILTER)
    String filter = "(!(" + WAMP_TOPIC_EVENT_PROPERTY + "=*))";

    private Map<String, WampClient> clients = Collections.unmodifiableMap(new HashMap<String, WampClient>());

    private Map<String, ExportedService> registry = Collections.unmodifiableMap(new HashMap<String, ExportedService>());

    @SuppressWarnings("UnusedDeclaration")
    public WampController(@Requires WisdomEngine engine) {
        String urlPrefix = "http://" + engine.hostname();
        if (engine.httpPort() != 80) {
            urlPrefix += ":" + engine.httpPort();
        }
        prefix = urlPrefix + Constants.WAMP_ROUTE;
        errorPrefix = prefix + Constants.WAMP_ERROR;
    }

    /**
     * Constructor used for testing purpose only.
     * Do not use directly outside the test scope.
     *
     * @param json           the json service
     * @param publisher      the publisher
     * @param prefix        the  url prefix
     */
    public WampController(Json json, Publisher publisher, String prefix) {
        this.json = json;
        this.publisher = publisher;
        this.prefix = prefix;
        this.errorPrefix = prefix + Constants.WAMP_ERROR;
    }

    @Opened(Constants.WAMP_ROUTE)
    public void open(@Parameter("client") String id) {
        LOGGER.info("Opening Wamp connection for client " + id);
        WampClient client = addClient(id);
        if (client != null) {
            // Send the welcome message.
            sendOnWebSocket(new Welcome(client.session()).toJson(json), client);
        }
    }

    @Closed(Constants.WAMP_ROUTE)
    public void close(@Parameter("client") String id) {
        removeClient(id);
    }

    @OnMessage(Constants.WAMP_ROUTE)
    public void onMessage(@Parameter("client") String id, @Body ArrayNode message) {
        MessageType type = MessageType.getType(message.get(0).asInt());
        Map.Entry<String, WampClient> entry = getClientById(id);

        switch (type) {
            case PREFIX:
                handlePrefixMessage(id, entry, message);
                break;
            case CALL:
                handleCallMessage(id, entry, message);
                break;
            case SUBSCRIBE:
                handleSubscription(id, entry, message);
                break;
            case UNSUBSCRIBE:
                handleUnsubscription(id, entry, message);
                break;
            case PUBLISH:
                handlePublication(id, entry, message);
                break;
            default:
                LOGGER.error("Illegal WAMP message type code {}", type.code());
                break;
        }
    }

    private void handleCallMessage(String id, Map.Entry<String, WampClient> entry, ArrayNode message) {
        if (entry == null) {
            LOGGER.error("Invalid CALL message, cannot identify the client {}", id);
            return;
        }
        if (message.get(1) == null || message.get(1).asText() == null) {
            LOGGER.error("Invalid CALL message, callId not defined in the CALL message {}", message.toString());
            sendOnWebSocket(<commit_msg>Fix #100<commit_after>/*
 * #%L
 * Wisdom-Framework
 * %%
 * Copyright (C) 2013 - 2014 Wisdom Framework
 * %%
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * 
 *      http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 * #L%
 */
package org.wisdom.wamp;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ArrayNode;
import org.apache.felix.ipojo.annotations.*;
import org.osgi.service.event.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.wisdom.api.Controller;
import org.wisdom.api.DefaultController;
import org.wisdom.api.annotations.*;
import org.wisdom.api.content.Json;
import org.wisdom.api.engine.WisdomEngine;
import org.wisdom.api.http.websockets.Publisher;
import org.wisdom.wamp.messages.*;
import org.wisdom.wamp.messages.Event;
import org.wisdom.wamp.services.ExportedService;
import org.wisdom.wamp.services.RegistryException;
import org.wisdom.wamp.services.Wamp;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.*;

/**
 * The WAMP main controller.
 * This controller is responsible for receiving the WAMP messages and sending the responses.
 * Components willing to be exposed using the Wamp protocol must register themselves using the WampRegistry service.
 */
@Component(immediate = true)
@Provides(specifications = {Wamp.class, EventHandler.class, Controller.class})
@Instantiate
public class WampController extends DefaultController implements Wamp, EventHandler {

    private static final Logger LOGGER = LoggerFactory.getLogger(WampController.class);

    private final String prefix;
    private final String errorPrefix;

    @Requires
    Publisher publisher;

    @Requires
    Json json;

    @Requires(optional = true)
    EventAdmin ea;

    /**
     * The topics from the event admin listened by this component to transfer the event to WAMP.
     */
    @ServiceProperty(name=EventConstants.EVENT_TOPIC)
    String[] topics = new String[] {
            "wamp/*"
    };

    /**
     * To avoid loop, we ignore all event send by us. The event we sent have a 'wamp.topic' property,
     * to we accept only event without this property.
     */
    @ServiceProperty(name=EventConstants.EVENT_FILTER)
    String filter = "(!(" + WAMP_TOPIC_EVENT_PROPERTY + "=*))";

    private Map<String, WampClient> clients = Collections.unmodifiableMap(new HashMap<String, WampClient>());

    private Map<String, ExportedService> registry = Collections.unmodifiableMap(new HashMap<String, ExportedService>());

    @SuppressWarnings("UnusedDeclaration")
    public WampController(@Requires WisdomEngine engine) {
        String urlPrefix = "http://" + engine.hostname();
        if (engine.httpPort() != 80) {
            urlPrefix += ":" + engine.httpPort();
        }
        prefix = urlPrefix + Constants.WAMP_ROUTE;
        errorPrefix = prefix + Constants.WAMP_ERROR;
    }

    /**
     * Constructor used for testing purpose only.
     * Do not use directly outside the test scope.
     *
     * @param json           the json service
     * @param publisher      the publisher
     * @param prefix        the  url prefix
     */
    public WampController(Json json, Publisher publisher, String prefix) {
        this.json = json;
        this.publisher = publisher;
        this.prefix = prefix;
        this.errorPrefix = prefix + Constants.WAMP_ERROR;
    }

    @Opened(Constants.WAMP_ROUTE)
    public void open(@Parameter("client") String id) {
        LOGGER.info("Opening Wamp connection for client " + id);
        WampClient client = addClient(id);
        if (client != null) {
            // Send the welcome message.
            sendOnWebSocket(new Welcome(client.session()).toJson(json), client);
        }
    }

    @Closed(Constants.WAMP_ROUTE)
    public void close(@Parameter("client") String id) {
        removeClient(id);
    }

    @OnMessage(Constants.WAMP_ROUTE)
    public void onMessage(@Parameter("client") String id, @Body ArrayNode message) {
        MessageType type = MessageType.getType(message.get(0).asInt());
        Map.Entry<String, WampClient> entry = getClientById(id);

        switch (type) {
            case PREFIX:
                handlePrefixMessage(id, entry, message);
                break;
            case CALL:
                handleCallMessage(id, entry, message);
                break;
            case SUBSCRIBE:
                handleSubscription(id, entry, message);
                break;
            case UNSUBSCRIBE:
                handleUnsubscription(id, entry, message);
                break;
            case PUBLISH:
                handlePublication(id, entry, message);
                break;
            default:
                LOGGER.error("Illegal WAMP message type code {}", type.code());
                break;
        }
    }

    @Invalidate
    public void stop() {
        registry.clear();
        clients.clear();
    }

    private void handleCallMessage(String id, Map.Entry<String, WampClient> entry, ArrayNode message) {
        if (entry == null) {
            LOGGER.error("Invalid CALL message, cannot identify the client {}", id);
            return;
        }
        if (message.get(1) == null || message.get(1).asText() == null) {
            LOGGER.error("Invalid CALL message, callId not defined in the CALL message {}", message.toString());
            sendOnWebSocket(
{'fraction_non_alphanumeric': 0.10118740320082602, 'fraction_numerical': 0.0035277921183961454, 'mean_word_length': 2.7996077149395227, 'pattern_counts': {'":': 2, '<': 15, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 4}, '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': '7248454', 'n_tokens_mistral': 3287, 'n_tokens_neox': 3185, 'n_words': 924}