mirror of https://github.com/dtm-labs/dtm.git
yedf2
3 years ago
committed by
GitHub
GitHub
36 changed files with 10374 additions and 971 deletions
File diff suppressed because it is too large
@ -0,0 +1,155 @@ |
|||
<template> |
|||
<div> |
|||
<a-button type="primary" class="mb-2" @click="handleTopicSubscribe('')">Subscribe</a-button> |
|||
<a-table :columns="columns" :data-source="dataSource" :loading="loading" :pagination="false"> |
|||
<template #bodyCell="{column, record}"> |
|||
<template v-if="column.key === 'subscribers'"> |
|||
<span>{{ JSON.parse(record.v).length }}</span> |
|||
</template> |
|||
<template v-if="column.key === 'action'"> |
|||
<span> |
|||
<a class="mr-2 font-medium" @click="handleTopicSubscribe(record.k)">Subscribe</a> |
|||
<a class="mr-2 font-medium" @click="handleTopicDetail(record.k,record.v)">Detail</a> |
|||
<a class="text-red-400 font-medium" @click="handleDeleteTopic(record.k)">Delete</a> |
|||
</span> |
|||
</template> |
|||
</template> |
|||
</a-table> |
|||
<div class="flex justify-center mt-2 text-lg pager" v-if="canPrev || canNext"> |
|||
<a-button type="text" :disabled="!canPrev" @click="handlePrevPage">Previous</a-button> |
|||
<a-button type="text" :disabled="!canNext" @click="handleNextPage">Next</a-button> |
|||
</div> |
|||
|
|||
<DialogTopicDetail ref="topicDetail" @unsubscribed="handleRefreshData"/> |
|||
<DialogTopicSubscribe ref="topicSubscribe" @subscribed="handleRefreshData"/> |
|||
</div> |
|||
</template> |
|||
<script setup lang="ts"> |
|||
import {deleteTopic, IListAllKVReq, listKVPairs} from '/@/api/api_dtm' |
|||
import {computed, ref} from 'vue-demi' |
|||
import {usePagination} from 'vue-request' |
|||
import DialogTopicDetail from './_Components/DialogTopicDetail.vue'; |
|||
import DialogTopicSubscribe from './_Components/DialogTopicSubscribe.vue'; |
|||
import {message, Modal} from 'ant-design-vue'; |
|||
|
|||
const columns = [ |
|||
{ |
|||
title: 'Name', |
|||
dataIndex: 'k', |
|||
key: 'name' |
|||
}, { |
|||
title: 'Subscribers', |
|||
dataIndex: 'v', |
|||
key: 'subscribers' |
|||
}, { |
|||
title: 'Version', |
|||
dataIndex: 'version', |
|||
key: 'version' |
|||
}, { |
|||
title: 'Action', |
|||
key: 'action' |
|||
} |
|||
] |
|||
|
|||
const pages = ref(['']) |
|||
const curPage = ref(1) |
|||
|
|||
const canPrev = computed(() => { |
|||
return curPage.value > 1 |
|||
}) |
|||
|
|||
const canNext = computed(() => { |
|||
return data.value?.data.next_position !== "" |
|||
}) |
|||
|
|||
type Data = { |
|||
kv: { |
|||
k: string |
|||
v: string |
|||
version: number |
|||
}[] |
|||
next_position: string |
|||
} |
|||
|
|||
const queryData = (params: IListAllKVReq) => { |
|||
return listKVPairs<Data>(params) |
|||
} |
|||
|
|||
const {data, run, current, loading, pageSize} = usePagination(queryData, { |
|||
defaultParams: [ |
|||
{ |
|||
cat: "topics", |
|||
limit: 10, |
|||
} |
|||
], |
|||
pagination: { |
|||
pageSizeKey: 'limit' |
|||
} |
|||
}) |
|||
|
|||
const dataSource = computed(() => data.value?.data.kv || []) |
|||
|
|||
const handlePrevPage = () => { |
|||
curPage.value -= 1; |
|||
const params = { |
|||
cat: "topics", |
|||
limit: pageSize.value, |
|||
position: pages.value[curPage.value] as string |
|||
} |
|||
run(params) |
|||
} |
|||
|
|||
const handleNextPage = () => { |
|||
curPage.value += 1; |
|||
pages.value[curPage.value] = data.value?.data.next_position as string |
|||
|
|||
run({ |
|||
cat: "topics", |
|||
position: data.value?.data.next_position, |
|||
limit: pageSize.value, |
|||
}) |
|||
} |
|||
|
|||
const handleRefreshData = () => { |
|||
run({cat: 'topics', limit: pageSize.value}) |
|||
} |
|||
|
|||
const handleDeleteTopic = (topic: string) => { |
|||
Modal.confirm({ |
|||
title: 'Delete', |
|||
content: 'Do you want delete this topic? ', |
|||
okText: 'Yes', |
|||
okType: 'danger', |
|||
cancelText: 'Cancel', |
|||
onOk: async () => { |
|||
await deleteTopic(topic) |
|||
message.success('Delete topic succeed') |
|||
run({cat: 'topics', limit: pageSize.value}) |
|||
} |
|||
}) |
|||
} |
|||
|
|||
const topicDetail = ref<null | { open: (topic: string, subscribers: string) => null }>(null) |
|||
const handleTopicDetail = (topic: string, subscribers: string) => { |
|||
topicDetail.value?.open(topic, subscribers) |
|||
} |
|||
|
|||
const topicSubscribe = ref<null | { open: (topic: string) => null }>(null) |
|||
const handleTopicSubscribe = (topic: string) => { |
|||
topicSubscribe.value?.open(topic) |
|||
} |
|||
</script> |
|||
|
|||
<style lang="postcss" scoped> |
|||
::deep .ant-pagination-item { |
|||
display: none; |
|||
} |
|||
|
|||
.pager .ant-btn-text { |
|||
font-weight: 500; |
|||
} |
|||
|
|||
.pager .ant-btn { |
|||
padding: 6px; |
|||
} |
|||
</style> |
|||
@ -0,0 +1,99 @@ |
|||
<template> |
|||
<div> |
|||
<a-modal v-model:visible="visible" :title=topicName width="100%" wrap-class-name="full-modal" :footer="null"> |
|||
<a-table :columns="columns" :data-source="dataSource" :pagination="false"> |
|||
<template #bodyCell="{column, record}"> |
|||
<template v-if="column.key === 'action'"> |
|||
<span> |
|||
<a class="text-red-400 font-medium" @click="handleUnsubscribe(record.url)">Unsubscribe</a> |
|||
</span> |
|||
</template> |
|||
</template> |
|||
</a-table> |
|||
<!-- <div class="mt-10 relative"> |
|||
<a-textarea id="qs" v-model:value="textVal" :auto-size="{ minRows: 10, maxRows: 10 }" /> |
|||
<screenfull class="absolute right-2 top-3 z-50" identity="qs" /> |
|||
</div> --> |
|||
</a-modal> |
|||
</div> |
|||
</template> |
|||
|
|||
<script setup lang="ts"> |
|||
import {ref} from 'vue'; |
|||
import {unsubscribe} from "/@/api/api_dtm"; |
|||
import {message, Modal} from "ant-design-vue"; |
|||
// import VueJsonPretty from 'vue-json-pretty'; |
|||
// import 'vue-json-pretty/lib/styles.css' |
|||
|
|||
const dataSource = ref<Subscriber[]>([]) |
|||
const visible = ref(false) |
|||
const topicName = ref<string>(""); |
|||
|
|||
const open = async (topic: string, subscribers: string) => { |
|||
dataSource.value = JSON.parse(subscribers) |
|||
topicName.value = topic |
|||
visible.value = true |
|||
} |
|||
|
|||
const columns = [ |
|||
{ |
|||
title: 'URL', |
|||
dataIndex: 'url', |
|||
key: 'url' |
|||
}, { |
|||
title: 'Remark', |
|||
dataIndex: 'remark', |
|||
key: 'remark' |
|||
}, { |
|||
title: 'Action', |
|||
key: 'action' |
|||
} |
|||
] |
|||
|
|||
interface Subscriber { |
|||
url: string |
|||
remark: string |
|||
} |
|||
|
|||
const handleUnsubscribe = async (url: string) => { |
|||
Modal.confirm({ |
|||
title: 'Unsubscribe', |
|||
content: 'Do you want unsubscribe this topic?', |
|||
okText: 'Yes', |
|||
okType: 'danger', |
|||
cancelText: 'Cancel', |
|||
onOk: async () => { |
|||
await unsubscribe({ |
|||
topic: topicName.value, |
|||
url: url |
|||
}) |
|||
message.success('Unsubscribe topic succeed') |
|||
location.reload() |
|||
} |
|||
}) |
|||
} |
|||
|
|||
defineExpose({ |
|||
open |
|||
}) |
|||
|
|||
</script> |
|||
|
|||
<style lang="postcss"> |
|||
.full-modal .ant-modal { |
|||
max-width: 100%; |
|||
top: 0; |
|||
padding-bottom: 0; |
|||
margin: 0; |
|||
} |
|||
|
|||
.full-modal .ant-modal-content { |
|||
display: flex; |
|||
flex-direction: column; |
|||
height: calc(100vh); |
|||
} |
|||
|
|||
.full-modal .ant-modal-body { |
|||
flex: 1; |
|||
} |
|||
</style> |
|||
@ -0,0 +1,71 @@ |
|||
<template> |
|||
<div> |
|||
<a-modal v-model:visible="visible" width="60%" title="Topic Subscribe" :confirm-loading="confirmLoading" |
|||
@ok="handleSubscribe"> |
|||
<a-form v-bind="layout" :mode="form"> |
|||
<a-form-item label="Topic: "> |
|||
<a-input v-model:value="form.topic" placeholder="Please input your topic..."/> |
|||
</a-form-item> |
|||
<a-form-item label="URL: "> |
|||
<a-input v-model:value="form.url" placeholder="Please input your url..."/> |
|||
</a-form-item> |
|||
<a-form-item label="Remark"> |
|||
<a-textarea v-model:value="form.remark" :rows="6" placeholder="Please input your remark..."/> |
|||
</a-form-item> |
|||
</a-form> |
|||
</a-modal> |
|||
</div> |
|||
</template> |
|||
|
|||
<script setup lang="ts"> |
|||
import {message} from 'ant-design-vue'; |
|||
import {reactive, ref} from 'vue'; |
|||
import {subscribe} from '/@/api/api_dtm' |
|||
|
|||
interface formState { |
|||
topic: string |
|||
url: string |
|||
remark: string |
|||
} |
|||
|
|||
const layout = { |
|||
labelCol: {span: 4}, |
|||
wrapperCol: {span: 16}, |
|||
} |
|||
|
|||
const form = reactive<formState>({ |
|||
topic: '', |
|||
url: '', |
|||
remark: '' |
|||
}) |
|||
|
|||
const visible = ref(false) |
|||
const open = async (topic: string) => { |
|||
form.topic = topic |
|||
visible.value = true |
|||
} |
|||
|
|||
const emit = defineEmits(['subscribed']) |
|||
|
|||
const confirmLoading = ref<boolean>(false) |
|||
const handleSubscribe = async () => { |
|||
confirmLoading.value = true |
|||
await subscribe<string>(form).then( |
|||
() => { |
|||
visible.value = false |
|||
message.success('Subscribe succeed') |
|||
confirmLoading.value = false |
|||
emit('subscribed') |
|||
} |
|||
).catch(() => { |
|||
message.error('Failed') |
|||
confirmLoading.value = false |
|||
return |
|||
}) |
|||
} |
|||
|
|||
defineExpose({ |
|||
open |
|||
}) |
|||
|
|||
</script> |
|||
File diff suppressed because it is too large
@ -0,0 +1,115 @@ |
|||
package dtmsvr |
|||
|
|||
import ( |
|||
"errors" |
|||
|
|||
"github.com/dtm-labs/dtm/client/dtmcli/dtmimp" |
|||
"github.com/dtm-labs/dtm/client/dtmcli/logger" |
|||
) |
|||
|
|||
const ( |
|||
topicsCat = "topics" |
|||
) |
|||
|
|||
var topicsMap = map[string]Topic{} |
|||
|
|||
//Topic define topic info
|
|||
type Topic struct { |
|||
Name string `json:"k"` |
|||
Subscribers []Subscriber `json:"v"` |
|||
Version uint64 `json:"version"` |
|||
} |
|||
|
|||
//Subscriber define subscriber info
|
|||
type Subscriber struct { |
|||
URL string `json:"url"` |
|||
Remark string `json:"remark"` |
|||
} |
|||
|
|||
func topic2urls(topic string) []string { |
|||
urls := make([]string, len(topicsMap[topic].Subscribers)) |
|||
for k, subscriber := range topicsMap[topic].Subscribers { |
|||
urls[k] = subscriber.URL |
|||
} |
|||
return urls |
|||
} |
|||
|
|||
// Subscribe subscribes topic, create topic if not exist
|
|||
func Subscribe(topic, url, remark string) error { |
|||
if topic == "" { |
|||
return errors.New("empty topic") |
|||
} |
|||
if url == "" { |
|||
return errors.New("empty url") |
|||
} |
|||
|
|||
newSubscriber := Subscriber{ |
|||
URL: url, |
|||
Remark: remark, |
|||
} |
|||
kvs := GetStore().FindKV(topicsCat, topic) |
|||
if len(kvs) == 0 { |
|||
return GetStore().CreateKV(topicsCat, topic, dtmimp.MustMarshalString([]Subscriber{newSubscriber})) |
|||
} |
|||
|
|||
subscribers := []Subscriber{} |
|||
dtmimp.MustUnmarshalString(kvs[0].V, &subscribers) |
|||
for _, subscriber := range subscribers { |
|||
if subscriber.URL == url { |
|||
return errors.New("this url exists") |
|||
} |
|||
} |
|||
subscribers = append(subscribers, newSubscriber) |
|||
kvs[0].V = dtmimp.MustMarshalString(subscribers) |
|||
return GetStore().UpdateKV(&kvs[0]) |
|||
} |
|||
|
|||
// Unsubscribe unsubscribes the topic
|
|||
func Unsubscribe(topic, url string) error { |
|||
if topic == "" { |
|||
return errors.New("empty topic") |
|||
} |
|||
if url == "" { |
|||
return errors.New("empty url") |
|||
} |
|||
|
|||
kvs := GetStore().FindKV(topicsCat, topic) |
|||
if len(kvs) == 0 { |
|||
return errors.New("no such a topic") |
|||
} |
|||
subscribers := []Subscriber{} |
|||
dtmimp.MustUnmarshalString(kvs[0].V, &subscribers) |
|||
if len(subscribers) == 0 { |
|||
return errors.New("this topic is empty") |
|||
} |
|||
n := len(subscribers) |
|||
for k, subscriber := range subscribers { |
|||
if subscriber.URL == url { |
|||
subscribers = append(subscribers[:k], subscribers[k+1:]...) |
|||
break |
|||
} |
|||
} |
|||
if len(subscribers) == n { |
|||
return errors.New("no such an url ") |
|||
} |
|||
kvs[0].V = dtmimp.MustMarshalString(subscribers) |
|||
return GetStore().UpdateKV(&kvs[0]) |
|||
} |
|||
|
|||
// updateTopicsMap updates the topicsMap variable, unsafe for concurrent
|
|||
func updateTopicsMap() { |
|||
kvs := GetStore().FindKV(topicsCat, "") |
|||
for _, kv := range kvs { |
|||
topic := topicsMap[kv.K] |
|||
if topic.Version >= kv.Version { |
|||
continue |
|||
} |
|||
newTopic := Topic{} |
|||
newTopic.Name = kv.K |
|||
newTopic.Version = kv.Version |
|||
dtmimp.MustUnmarshalString(kv.V, &newTopic.Subscribers) |
|||
topicsMap[kv.K] = newTopic |
|||
logger.Infof("topic updated. old topic:%v new topic:%v", topicsMap[kv.K], newTopic) |
|||
} |
|||
logger.Infof("all topic updated. topic:%v", topicsMap) |
|||
} |
|||
@ -0,0 +1,151 @@ |
|||
package test |
|||
|
|||
import ( |
|||
"context" |
|||
"strconv" |
|||
"sync" |
|||
"testing" |
|||
|
|||
"github.com/dtm-labs/dtm/client/dtmcli" |
|||
"github.com/dtm-labs/dtm/client/dtmcli/dtmimp" |
|||
"github.com/dtm-labs/dtm/client/dtmgrpc/dtmgimp" |
|||
"github.com/dtm-labs/dtm/client/dtmgrpc/dtmgpb" |
|||
"github.com/dtm-labs/dtm/dtmutil" |
|||
"github.com/pkg/errors" |
|||
"github.com/stretchr/testify/assert" |
|||
) |
|||
|
|||
const ( |
|||
testTopicTestTopicNormal = "test_topic_TestTopicNormal" |
|||
testTopicTestConcurrentUpdateTopic = "concurrent_topic_TestConcurrentUpdateTopic" |
|||
) |
|||
|
|||
func TestTopicNormal(t *testing.T) { |
|||
testSubscribe(t, httpSubscribe) |
|||
testUnsubscribe(t, httpUnsubscribe) |
|||
testDeleteTopic(t, httpDeleteTopic) |
|||
|
|||
testSubscribe(t, grpcSubscribe) |
|||
testUnsubscribe(t, grpcUnsubscribe) |
|||
testDeleteTopic(t, grpcDeleteTopic) |
|||
} |
|||
|
|||
func TestConcurrentUpdateTopic(t *testing.T) { |
|||
var wg sync.WaitGroup |
|||
var urls []string |
|||
var errNum int |
|||
concurrentTimes := 20 |
|||
// concurrently updates the topic, part of them succeed
|
|||
for i := 0; i < concurrentTimes; i++ { |
|||
wg.Add(1) |
|||
go func(i int) { |
|||
url := "http://dtm/test" + strconv.Itoa(i) |
|||
err := httpSubscribe(testTopicTestConcurrentUpdateTopic, url) |
|||
if err == nil { |
|||
urls = append(urls, url) |
|||
} else { |
|||
errNum++ |
|||
} |
|||
wg.Done() |
|||
}(i) |
|||
} |
|||
wg.Wait() |
|||
assert.True(t, len(urls) > 0) |
|||
|
|||
// delete successfully subscribed urls above, all of them should succeed
|
|||
for _, url := range urls { |
|||
assert.Nil(t, httpUnsubscribe(testTopicTestConcurrentUpdateTopic, url)) |
|||
} |
|||
|
|||
// finally, the topic version should be correct
|
|||
m := map[string]interface{}{} |
|||
resp, err := dtmcli.GetRestyClient().R().SetQueryParams(map[string]string{ |
|||
"cat": "topics", |
|||
"key": testTopicTestConcurrentUpdateTopic, |
|||
}).Get(dtmutil.DefaultHTTPServer + "/queryKV") |
|||
assert.Nil(t, err) |
|||
dtmimp.MustUnmarshalString(resp.String(), &m) |
|||
dtmimp.MustRemarshal(m["kv"].([]interface{})[0], &m) |
|||
assert.Equal(t, float64((concurrentTimes-errNum)*2), m["version"]) |
|||
} |
|||
|
|||
func testSubscribe(t *testing.T, subscribe func(topic, url string) error) { |
|||
assert.Nil(t, subscribe(testTopicTestTopicNormal, "http://dtm/test1")) |
|||
assert.Error(t, subscribe(testTopicTestTopicNormal, "http://dtm/test1")) // error:repeat subscription
|
|||
assert.Error(t, subscribe("", "http://dtm/test1")) // error:empty topic
|
|||
assert.Error(t, subscribe(testTopicTestTopicNormal, "")) // error:empty url
|
|||
assert.Nil(t, subscribe(testTopicTestTopicNormal, "http://dtm/test2")) |
|||
} |
|||
|
|||
func testUnsubscribe(t *testing.T, unsubscribe func(topic, url string) error) { |
|||
assert.Nil(t, unsubscribe(testTopicTestTopicNormal, "http://dtm/test1")) |
|||
assert.Error(t, unsubscribe(testTopicTestTopicNormal, "http://dtm/test1")) // error:repeat unsubscription
|
|||
assert.Error(t, unsubscribe("", "http://dtm/test1")) // error:empty topic
|
|||
assert.Error(t, unsubscribe(testTopicTestTopicNormal, "")) // error:empty url
|
|||
assert.Error(t, unsubscribe("non_existent_topic", "http://dtm/test1")) // error:unsubscribe a non-existent topic
|
|||
assert.Nil(t, unsubscribe(testTopicTestTopicNormal, "http://dtm/test2")) |
|||
assert.Error(t, unsubscribe(testTopicTestTopicNormal, "http://dtm/test2")) |
|||
} |
|||
|
|||
func testDeleteTopic(t *testing.T, deleteTopic func(topic string) error) { |
|||
assert.Error(t, deleteTopic("non_existent_testDeleteTopic")) |
|||
assert.Nil(t, deleteTopic(testTopicTestTopicNormal)) |
|||
} |
|||
|
|||
func httpSubscribe(topic, url string) error { |
|||
resp, err := dtmcli.GetRestyClient().R().SetQueryParams(map[string]string{ |
|||
"topic": topic, |
|||
"url": url, |
|||
"remark": "for test", |
|||
}).Get(dtmutil.DefaultHTTPServer + "/subscribe") |
|||
e2p(err) |
|||
if resp.StatusCode() != 200 { |
|||
err = errors.Errorf("Http Request Error. Resp:%v", resp.String()) |
|||
} |
|||
return err |
|||
} |
|||
|
|||
func httpUnsubscribe(topic, url string) error { |
|||
resp, err := dtmcli.GetRestyClient().R().SetQueryParams(map[string]string{ |
|||
"topic": topic, |
|||
"url": url, |
|||
}).Get(dtmutil.DefaultHTTPServer + "/unsubscribe") |
|||
e2p(err) |
|||
if resp.StatusCode() != 200 { |
|||
err = errors.Errorf("Http Request Error. Resp:%+v", resp.String()) |
|||
} |
|||
return err |
|||
} |
|||
|
|||
func httpDeleteTopic(topic string) error { |
|||
resp, err := dtmcli.GetRestyClient().R().Delete(dtmutil.DefaultHTTPServer + "/topic/" + topic) |
|||
e2p(err) |
|||
if resp.StatusCode() != 200 { |
|||
err = errors.Errorf("Http Request Error. Resp:%+v", resp.String()) |
|||
} |
|||
return err |
|||
} |
|||
|
|||
func grpcSubscribe(topic, url string) error { |
|||
_, err := dtmgimp.MustGetDtmClient(dtmutil.DefaultGrpcServer).Subscribe(context.Background(), |
|||
&dtmgpb.DtmTopicRequest{ |
|||
Topic: topic, |
|||
URL: url, |
|||
Remark: "for test"}) |
|||
return err |
|||
} |
|||
|
|||
func grpcUnsubscribe(topic, url string) error { |
|||
_, err := dtmgimp.MustGetDtmClient(dtmutil.DefaultGrpcServer).Unsubscribe(context.Background(), |
|||
&dtmgpb.DtmTopicRequest{ |
|||
Topic: topic, |
|||
URL: url}) |
|||
return err |
|||
} |
|||
|
|||
func grpcDeleteTopic(topic string) error { |
|||
_, err := dtmgimp.MustGetDtmClient(dtmutil.DefaultGrpcServer).DeleteTopic(context.Background(), |
|||
&dtmgpb.DtmTopicRequest{ |
|||
Topic: topic}) |
|||
return err |
|||
} |
|||
Loading…
Reference in new issue