<commit_before>package magina

import (
	"fmt"
	"github.com/ruizeng/magina/packets"
	"github.com/streadway/amqp"
	"log"
	"net"
)

const (
	defaultPubsubExchange = "pubsub"
	defaultCrpcExchange   = "crpc"
	defaultDrpcExchange   = "drpc"
)

type Client struct {
	Conn       net.Conn
	Broker     *Broker
	Identifier string
	Channel    *amqp.Channel
}

func (c *Client) initRabbit() error {
	var err error
	if c.Channel == nil {
		c.Channel, err = c.Broker.RabbitConnection.Channel()
		if err != nil {
			return err
		}

		err = c.Channel.ExchangeDeclare(
			defaultPubsubExchange, // name
			"topic",               // type
			true,                  // durable
			false,                 // auto-deleted
			false,                 // internal
			false,                 // no-wait
			nil,                   // arguments
		)

		if err != nil {
			return err
		}
	}

	return nil
}

func (c *Client) trySendPacket(packet packets.ControlPacket) {
	log.Printf("try send packet******** \n%v\n**********\n", packet)
	err := packet.Write(c.Conn)
	if err != nil {
		log.Println(err)
	}
}

func (c *Client) handlePublish(pub *packets.PublishPacket) error {
	if c.Channel == nil {
		return fmt.Errorf("client channel not ready")
	}
	err := c.Channel.Publish(defaultPubsubExchange,
		pub.TopicName, false, false,
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        pub.Payload,
		})
	return err
}

func (c *Client) handleSubscribe(sub *packets.SubscribePacket) error {
	if c.Channel == nil {
		return fmt.Errorf("client channel not ready")
	}
	q, err := c.Channel.QueueDeclare(
		"",    // name
		false, // durable
		false, // delete when usused
		true,  // exclusive
		false, // no-wait
		nil,   // arguments
	)
	if err != nil {
		return err
	}

	err = c.Channel.QueueBind(
		q.Name,                // queue name
		sub.Topics[0],         // routing key
		defaultPubsubExchange, // exchange
		false,
		nil,
	)
	if err != nil {
		return err
	}

	msgs, err := c.Channel.Consume(
		q.Name, // queue
		"",     // consumer
		true,   // auto ack
		false,  // exclusive
		false,  // no local
		false,  // no wait
		nil,    // args
	)
	if err != nil {
		return err
	}

	go func() {
		for d := range msgs {
			pub := packets.NewControlPacket(packets.Publish).(*packets.PublishPacket)
			pub.Payload = d.Body
			pub.TopicName = d.RoutingKey
			c.trySendPacket(pub)
		}
	}()

	suback := packets.NewControlPacket(packets.Suback).(*packets.SubackPacket)
	suback.MessageID = sub.MessageID
	c.trySendPacket(suback)
	return nil
}

func (c *Client) Serve() {
	defer func() {
		if c.Channel != nil {
			c.Channel.Close()
		}
		if c.Conn != nil {
			c.Conn.Close()
		}
	}()
	needDisconnect := false
	for {
		if needDisconnect {
			break
		}
		packet, err := packets.ReadPacket(c.Conn)
		if err != nil {
			log.Printf("reading packets from connection error: %v", err)
			break
		}
		log.Printf("packet received =========\n%v\n===============\n", packet)
		switch packet.(type) {
		case *packets.ConnectPacket:
			conn := packet.(*packets.ConnectPacket)
			ca := packets.NewControlPacket(packets.Connack).(*packets.ConnackPacket)
			ca.ReturnCode = conn.Validate()
			err = c.initRabbit()
			if err != nil {
				log.Printf("init rabbitmq for client failed: %v\n", err)
				ca.ReturnCode = packets.ErrRefusedServerUnavailable
			}
			c.trySendPacket(ca)
		case *packets.DisconnectPacket:
			log.Println("disconnecting client...")
			needDisconnect = true
		case *packets.PingreqPacket:
			pres := packets.NewControlPacket(packets.Pingresp)
			log.Println("ping back to cliend...")
			c.trySendPacket(pres)
		case *packets.PublishPacket:
			pub := packet.(*packets.PublishPacket)
			c.handlePublish(pub)
		case *packets.SubscribePacket:
			sub := packet.(*packets.SubscribePacket)
			c.handleSubscribe(sub)
		default:
			log.Println("unknown packet: ", packet)
		}
	}
}
<commit_msg>change: support topic unsubscribe bugfix: none<commit_after>package magina

import (
	"fmt"
	"github.com/ruizeng/magina/packets"
	"github.com/streadway/amqp"
	"log"
	"net"
)

const (
	defaultPubsubExchange = "pubsub"
	defaultCrpcExchange   = "crpc"
	defaultDrpcExchange   = "drpc"
)

type Client struct {
	Conn       net.Conn
	Broker     *Broker
	Identifier string
	Channel    *amqp.Channel
	TopicQueue map[string]string
}

func (c *Client) initRabbit() error {
	var err error
	if c.Channel == nil {
		c.Channel, err = c.Broker.RabbitConnection.Channel()
		if err != nil {
			return err
		}

		err = c.Channel.ExchangeDeclare(
			defaultPubsubExchange, // name
			"topic",               // type
			true,                  // durable
			false,                 // auto-deleted
			false,                 // internal
			false,                 // no-wait
			nil,                   // arguments
		)

		if err != nil {
			return err
		}
	}

	if c.TopicQueue == nil {
		c.TopicQueue = make(map[string]string)
	}

	return nil
}

func (c *Client) trySendPacket(packet packets.ControlPacket) error {
	log.Printf("try send packet******** \n%v\n**********\n", packet)
	return packet.Write(c.Conn)
}

func (c *Client) handlePublish(pub *packets.PublishPacket) error {
	if c.Channel == nil {
		return fmt.Errorf("client channel not ready")
	}
	err := c.Channel.Publish(defaultPubsubExchange,
		pub.TopicName, false, false,
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        pub.Payload,
		})
	return err
}

func (c *Client) handleSubscribe(sub *packets.SubscribePacket) error {
	if c.Channel == nil {
		return fmt.Errorf("client channel not ready")
	}
	q, err := c.Channel.QueueDeclare(
		"",    // name
		false, // durable
		true,  // delete when usused
		true,  // exclusive
		false, // no-wait
		nil,   // arguments
	)
	if err != nil {
		return err
	}

	topic := sub.Topics[0] // only suport one topic a time now

	err = c.Channel.QueueBind(
		q.Name, // queue name
		topic,  // routing key
		defaultPubsubExchange, // exchange
		false,
		nil,
	)
	if err != nil {
		return err
	}

	c.TopicQueue[topic] = q.Name

	msgs, err := c.Channel.Consume(
		q.Name, // queue
		"",     // consumer
		true,   // auto ack
		false,  // exclusive
		false,  // no local
		false,  // no wait
		nil,    // args
	)
	if err != nil {
		return err
	}

	go func() {
		for d := range msgs {
			pub := packets.NewControlPacket(packets.Publish).(*packets.PublishPacket)
			pub.Payload = d.Body
			pub.TopicName = d.RoutingKey
			c.trySendPacket(pub)
		}
	}()

	suback := packets.NewControlPacket(packets.Suback).(*packets.SubackPacket)
	suback.MessageID = sub.MessageID
	suback.Qos = sub.Qos
	c.trySendPacket(suback)
	return nil
}

func (c *Client) handleUnsubscribe(unsub *packets.UnsubscribePacket) error {
	if c.Channel == nil {
		return fmt.Errorf("client channel not ready")
	}

	topic := unsub.Topics[0] // only suport one topic a time now

	if queueName, exist := c.TopicQueue[topic]; exist {
		err := c.Channel.QueueUnbind(queueName, topic, defaultPubsubExchange, nil)
		delete(c.TopicQueue, topic)
		if err != nil {
			return err
		}
	}
	unsuback := packets.NewControlPacket(packets.Unsuback).(*packets.UnsubackPacket)
	unsuback.MessageID = unsub.MessageID
	unsuback.Qos = unsub.Qos
	return c.trySendPacket(unsuback)
}

func (c *Client) Serve() {
	defer func() {
		if c.Channel != nil {
			c.Channel.Close()
			c.Channel = nil
		}
		if c.Conn != nil {
			c.Conn.Close()
			c.Conn = nil
		}
	}()
	needDisconnect := false
	for {
		if needDisconnect {
			break
		}
		packet, err := packets.ReadPacket(c.Conn)
		if err != nil {
			log.Printf("reading packets from connection error: %v", err)
			break
		}
		log.Printf("packet received =========\n%v\n===============\n", packet)
		switch packet.(type) {
		case *packets.ConnectPacket:
			conn := packet.(*packets.ConnectPacket)
			ca := packets.NewControlPacket(packets.Connack).(*packets.ConnackPacket)
			ca.ReturnCode = conn.Validate()
			err = c.initRabbit()
			if err != nil {
				log.Printf("init rabbitmq for client failed: %v\n", err)
				ca.ReturnCode = packets.ErrRefusedServerUnavailable
			}
			err = c.trySendPacket(ca)
		case *packets.DisconnectPacket:
			log.Println("disconnecting client...")
			needDisconnect = true
		case *packets.PingreqPacket:
			pres := packets.NewControlPacket(packets.Pingresp)
			log.Println("ping back to cliend...")
			err = c.trySendPacket(pres)
		case *packets.PublishPacket:
			pub := packet.(*packets.PublishPacket)
			err = c.handlePublish(pub)
		case *packets.SubscribePacket:
			sub := packet.(*packets.SubscribePacket)
			err = c.handleSubscribe(sub)
		case *packets.UnsubscribePacket:
			unsub := packet.(*packets.UnsubscribePacket)
			err = c.handleUnsubscribe(unsub)
		default:
			log.Println("unknown packet: ", packet)
		}
		if err != nil {
			log.Println("handle packat error: ", err)
		}
	}
}

{'fraction_non_alphanumeric': 0.1409595786098706, 'fraction_numerical': 0.00034352456200618345, 'mean_word_length': 2.857773851590106, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '7091338', 'n_tokens_mistral': 3406, 'n_tokens_neox': 3067, 'n_words': 687}